实际生产环境Apache RocketMQ消息体过大的解决方案

前言

官方定义消息体默认大小为 4MB，普通顺序消息类型。事务、定时、延时类消息默认大小为64KB。如果超过限制则会抛出异常！

但实际工作中，需要使用到MQ进行异步解耦，传输的业务消息偶尔会遇到超过4MB，尤其在业务复杂的系统中，那么我们应该如何处理呢？

在我工作实际应用中，有以下几种解决方案。

解决方案

方案一：消息压缩

通常我们都是传递json消息数据，然后底层使用字节流进行传输。如果此时json数据超过4MB，则可以考虑进行消息压缩。

原理其实很好理解，比如我们经常使用的压缩包，可以把大文件进行压缩，依次减小文件大小。

那么我们这里需要使用到的就是字符压缩，把json字符串进行压缩，然后进行传输，原理图如下：

经过测试：我们原来5MB的数据可以压缩到230KB，1MB都不到，当然效果和数据以及压缩算法有关。如：大量重复字符则压缩效率就更高。

压缩解压代码如下：

@Slf4jpublicclassStringCompressUtils{/**
     * 使用gzip压缩字符串
     *
     * @param str 要压缩的字符串
     * @return 压缩结果字符
     */publicstaticStringcompress(String str){if(str ==null|| str.length()==0){return str;}ByteArrayOutputStream out =newByteArrayOutputStream();GZIPOutputStream gzip =null;try{
            gzip =newGZIPOutputStream(out);
            gzip.write(str.getBytes());}catch(IOException e){
            log.error("字符串压缩异常!", e);
            e.printStackTrace();}finally{IoUtil.close(gzip);}returnnewsun.misc.BASE64Encoder().encode(out.toByteArray());}/**
     * 使用gzip解压缩
     *
     * @param compressedStr 压缩字符串
     * @return 解压缩字符串
     */publicstaticStringuncompress(String compressedStr){if(compressedStr ==null){returnnull;}byte[] compressed =null;String decompressed =null;GZIPInputStream ginzip =null;ByteArrayInputStream in =null;ByteArrayOutputStream out =newByteArrayOutputStream();try{// 先解码
            compressed =newsun.misc.BASE64Decoder().decodeBuffer(compressedStr);
            in =newByteArrayInputStream(compressed);
            ginzip =newGZIPInputStream(in);byte[] buffer =newbyte[1024];int offset =-1;while((offset = ginzip.read(buffer))!=-1){
                out.write(buffer,0, offset);}

            decompressed = out.toString();}catch(IOException e){
            log.error("字符串解压缩异常!", e);
            e.printStackTrace();}finally{// 关流IoUtil.close(ginzip);IoUtil.close(in);IoUtil.close(out);}return decompressed;}

压缩流程：原始字符 -> 压缩 -> 压缩字符 -> 编码
解压流程：压缩字符 -> 解码 -> 解压 -> 原始字符

方案二：消息分割

方案一基本可以解决遇到的99%消息体过大的问题，如果不行则可以使用消息分割方案。

简而言之，就是把一个大消息体，进行分割成多个小消息体进行传输。运用了化整为零的思想，至于实现方案，有很多，我简单举个例子。

1. 一个大消息分割成多个小消息
2. 多个小消息拥有相同的消息标识，如UUID
3. 分割后小消息需要有一些元数据来标识自己，如 消息标识、一共分割了多少个、自己是第几个。
4. 传输后，消费者消费，然后根据元数据进行数据聚合还原。
5. 将还原后的消息走正常消费流程即可

方案三：OSS存储

方案一方案二几乎已经解决99.99%的场景了，如果还是不够，那就要实施核打击了。终极方案，OSS存储！此方法绝对可以解决100%的场景！

大消息 -> 写入到文件 -> 上传到文件服务器 -> 拿到URL -> 传输 -> 消费

生产者：大消息，写入文件，上传文件，拿到访问连接，发送访问连接给MQ
消费者：消费，拿到访问链接，读取文件，拿到消息，执行业务逻辑

快准狠！短平快！就是增加了中间件，其他一点毛病没有，效率更高！