RocketMQ结合源码告诉你消息量大为啥不需要手动压缩消息

背景

最近同事发现线上发送的RocketMQ消息太大，同事为了节省网络带宽和存储空间，手动压缩消息然后再进行消息发送，发现磁盘也没有明显的缩减

所以我打算结合源码告诉他RocketMQ自带的消息压缩

同时听了发出了这个表情包 alt text

RocketMQ版本

5.1.0

为什么需要压缩消息

首先说一下为什么需要消息压缩，原因其实很简单。就是为了节省网络带宽和存储空间。

在哪里压缩消息

我们的消息压缩可以在很多个地方进行

有两种方案

在client端进行压缩

比如我们可以在Producer发送消息的时候进行消息压缩

然后将压缩后的消息发送到Broker，broker只管存储

等到consumer需要消息的时候，原封不动的推送给消费者，由consumer自己进行解压缩

这种方式的好处是broker不需要关心消息的压缩和解压缩，只需要存储消息即可

在broker端进行压缩

这种方式就是Producer发送消息的时候，不进行压缩

RocketMQ在存储的时候自己进行消息压缩，consumer进行消息拉取的时候，broker进行消息解压缩，然后推送给consumer

这种方式就是消耗broker cpu，也不能节省网络带宽，只能节省存储空间

所以很明显是在client端进行压缩比较好

源码分析

这里我们来具体结合源码分析下

在消息发送org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendKernelImpl方法中会对消息进行压缩判断

alt text

tryToCompressMessage 消息压缩

什么消息会被压缩呢？

    private boolean tryToCompressMessage(final Message msg) {
        if (msg instanceof MessageBatch) {
            //batch does not support compressing right now
            return false;
        }
        byte[] body = msg.getBody();
        if (body != null) {
            if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {
                try {
                    byte[] data = compressor.compress(body, compressLevel);
                    if (data != null) {
                        msg.setBody(data);
                        return true;
                    }
                } catch (IOException e) {
                    log.error("tryToCompressMessage exception", e);
                    log.warn(msg.toString());
                }
            }
        }

        return false;
    }

批量消息不支持压缩
消息体长度大于defaultMQProducer.getCompressMsgBodyOverHowmuch()的时候进行压缩。默认1024 * 4 = 4kb
压缩算法是什么呢？

RocketMQ目前提供三种压缩算法

LZ4
ZSTD
ZLIB

默认压缩算法为ZLIB

    private CompressionType compressType = CompressionType.of(System.getProperty(MixAll.MESSAGE_COMPRESS_TYPE, "ZLIB"));

压缩等级为5

private int compressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));

消息压缩完后会通过sysFlag进行标记，表示消息进行了压缩，方便后续解压

                    sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
                    sysFlag |= compressType.getCompressionFlag();

消息解压

消息解压主要是在方法org.apache.rocketmq.common.message.MessageDecoder.decode(java.nio.ByteBuffer, boolean, boolean, boolean, boolean, boolean)中进行的

在client拉取到消息成功后对PullResult对象进行处理执行decodesBatch方法

alt text

消息解析decodesBatch方法会调用org.apache.rocketmq.common.message.MessageDecoder.decode(java.nio.ByteBuffer, boolean, boolean, boolean, boolean, boolean)方法

decode方法会对消息进行解压 alt text

总结

消息压缩主要是为了节省网络带宽和存储空间。
RocketMQ提供了三种压缩算法，分别是LZ4、ZSTD、ZLIB，默认为ZLIB
消息压缩主要是在Producer发送消息的时候进行压缩，broker只管存储
消息解压主要是在Consumer拉取消息的时候进行解压
RocketMQ消息压缩仅支持单条消息压缩，不支持批量消息压缩
一般消息压缩都会选择在client端进行压缩，这样可以节省broker的cpu

背景​

RocketMQ版本​

为什么需要压缩消息​

在哪里压缩消息​

源码分析​

tryToCompressMessage 消息压缩​

消息解压​

总结​

背景