public static byte[] encodeCacheMessageBody(INLONG_COMPRESSED_TYPE compressedType, List<ProxyEvent> events)
            throws IOException {
        // encode
        MessageObjs.Builder objs = MessageObjs.newBuilder();
        for (ProxyEvent event : events) {
            MessageObj.Builder builder = MessageObj.newBuilder();
            builder.setMsgTime(event.getMsgTime());
            builder.setSourceIp(event.getSourceIp());
            event.getHeaders().forEach((key, value) -> {
                builder.addParams(MapFieldEntry.newBuilder().setKey(key).setValue(value));
            });
            builder.setBody(ByteString.copyFrom(event.getBody()));
            objs.addMsgs(builder.build());
        }
        byte[] srcBytes = objs.build().toByteArray();
        // compress
        byte[] compressBytes = null;
        switch (compressedType) {
            case INLONG_SNAPPY:
                compressBytes = Snappy.compress(srcBytes);
                break;
            case INLONG_GZ:
                compressBytes = GzipUtils.compress(srcBytes);
                break;
            case INLONG_NO_COMPRESS:
            default:
                compressBytes = srcBytes;
                break;
        }
        return compressBytes;
    }
encodeCacheMessageBody

@param  compressedType
@param  events
@return byte array
@throws IOException
inlong-sdk/sdk-common/src/main/java/org/apache/inlong/sdk/commons/protocol/EventUtils.java