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

本文介绍: 官方定义消息体默认大小为 4MB，普通顺序消息类型。事务、定时、延时类消息默认大小为64KB。如果超过限制则会抛出异常！但实际工作中，需要使用到MQ进行异步解耦，传输的业务消息偶尔会遇到超过4MB，尤其在业务复杂的系统中，那么我们应该如何处理呢？在我工作实际应用中，有以下几种解决方案。

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

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

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

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

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

那么我们这里需要使用到的就是字符压缩，把js on 字符串进行压缩，然后进行传输，原理图如下：
压缩原理
经过测试：我们原来5MB的数据可以压缩到230KB，1MB都不到，当然效果和数据以及压缩算法有关。如：大量重复字符则压缩效率就更高。

@Slf4j
public class StringCompressUtils {

    /**
     * 使用gzip压缩字符串
     *
     * @param str 要压缩的字符串
     * @return 压缩结果字符
     */
    public static String compress(String str) {
        if (str == null || str.length() == 0) {
            return str;
        }
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        GZIPOutputStream gzip = null;
        try {
            gzip = new GZIPOutputStream(out);
            gzip.write(str.getBytes());
        } catch (IOException e) {
            log.error("字符串压缩异常!", e);
            e.printStackTrace();
        } finally {
            IoUtil.close(gzip);
        }
        return new sun.misc.BASE64Encoder().encode(out.toByteArray());
    }

    /**
     * 使用gzip解压缩
     *
     * @param compressedStr 压缩字符串
     * @return 解压缩字符串
     */
    public static String uncompress(String compressedStr) {
        if (compressedStr == null) {
            return null;
        }

        byte[] compressed = null;
        String decompressed = null;
        GZIPInputStream ginzip = null;
        ByteArrayInputStream in = null;
        ByteArrayOutputStream out = new ByteArrayOutputStream();

        try {
            // 先解码
            compressed = new sun.misc.BASE64Decoder().decodeBuffer(compressedStr);
            in = new ByteArrayInputStream(compressed);
            ginzip = new GZIPInputStream(in);
            byte[] buffer = new byte[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;
    }