当前位置：首页 > news >正文

【Alibaba中间件技术系列】「RocketMQ技术专题」RocketMQ消息发送的全部流程和落盘原理分析

news 2025/8/27 15:58:42

RocketMQ目前在国内应该是比较流行的MQ 了，目前本人也在公司的项目中进行使用和研究，借着这个机会，分析一下RocketMQ 发送一条消息到存储一条消息的过程，这样会对以后大家分析和研究RocketMQ相关的问题有一定的帮助。

分析的总体技术范围发送到存储，本文的主要目的是主要是为了认识一条消息并分析被发出且被存储的，代码中，关于 MQ 文件系统的优化，设计等。

来自官方源码example的一段发送代码：

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.start();
Message msg = new Message("TopicTest", "TagA", "OrderID188", "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
producer.shutdown();

直接看看send方法，send 方法会设置一个默认的 timeout：3秒。默认使用 SYNC 模式，另外有Async和OneWay模式。需要处理方法签名中的 Client 端的异常，网络异常，Broker 端的异常，线程中断异常。

DefaultMQProducerImpl 的 sendDefaultImpl方法就是发送的主要逻辑。

代码里，有个地方可以提一下，关于更新故障时间的策略，RocketMQ有一个类 MQFaultStrategy，用来处理MQ错误，然后对 MQ Server 进行服务降级。

如果发送一条消息在550ms以内，那么就不用降级，如果550毫秒以外，就进行容错降级（熔断）30 秒，以此类推。

再看DefaultMQProducerImpl 的 sendKernelImpl发送到内核的方法实现。

先找到broker的地址。尝试压缩大于4M 的消息（批量消息不压缩），然后执行各种钩子。

Request对象（存放数据）
Context 上下文对象（存放调用上下文）。

这里会设置一个消息生成时间，即bornTimestamp，后面使用消息轨迹的时候，可以查看。

默认情况下：如果采用SYNC 模式，就调用 MQClientAPIImpl 来发送消息，这一层还是在 Client 模块里，在这一层，会设置更详细的消息细节，构造命令对象。最后调用 remotingClient的 invokeSync 发送消息。

MQClientAPIImpl的sendMessage这一层，会给命令对象设置一个CmdCode，叫SEND_MESSAGE，这个东西就是一个和Broker的契约，Broker会根据这个Code进行不同的策略。

Netty 会使用 Handler 处理出去的数据和返回的数据，我们看看 Client 端 Netty 有哪些 Handler.

Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, false).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis()).option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize()).option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize()).handler(new ChannelInitializer() {public void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();if (nettyClientConfig.isUseTLS()) {if (null != sslContext) {pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc()));log.info("Prepend SSL handler");} else {log.warn("Connections are insecure as SSLContext is null!");}}pipeline.addLast(defaultEventExecutorGroup,new NettyEncoder(),new NettyDecoder(),new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),new NettyConnectManageHandler(),new NettyClientHandler());}});

使用了一个 Encoder，Decoder，空闲处理器，连接管理器，ClientHandler。

XXCoder就是对Cmd对象进行序列化和反序列化的，这里的空闲使用的读写最大空闲时间为120s，超过这个，就会触发空闲事件。

RocketMQ就会关闭Channel 连接。而针对空闲事件进行处理的就是连接管理器了。
连接管理器处理空闲、Close、Connect、异常等事件，使用监听器模式，不同的监听器对不同的事件进行处理。另外，这里也许可以借鉴 EventBus，每个事件可以设置多个监听器。

看了RocketMQ中 Netty 的设计，再看看返回值处理就简单了，NettyClientHandler 会在 channelRead0 方法处理 Netty Server 的返回值。对应 RMQ，则是 processMessageReceived 方法。该方法很简洁：

public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {final RemotingCommand cmd = msg;if (cmd != null) {switch (cmd.getType()) {case REQUEST_COMMAND:processRequestCommand(ctx, cmd);break;case RESPONSE_COMMAND:processResponseCommand(ctx, cmd);break;default:break;}}}

其实，这是一个模板方法，固定算法，由子类实现，分为 Request 实现和 Response 实现。我们看看 Response 实现。

public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {final int opaque = cmd.getOpaque();final ResponseFuture responseFuture = responseTable.get(opaque);if (responseFuture != null) {responseFuture.setResponseCommand(cmd);responseTable.remove(opaque);if (responseFuture.getInvokeCallback() != null) {executeInvokeCallback(responseFuture);} else {responseFuture.putResponse(cmd);responseFuture.release();}} else {log.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));log.warn(cmd.toString());}}

通过 cmd 对象的 Request ID 找到 Feature，执行 responseFuture.putResponse，设置返回值，唤醒阻塞等待的发送线程。

这里还有一个 release 调用，这个和异步发送有关，默认最大同时 65535 个异步请求，具体就不展开了。

到这里，唤醒阻塞的发送线程，返回数据，客户端层面的发送就结束了。

看源码，看到有个 SEND_MESSAGE Code，是 Client 和 Broker Server 的一个约定代码，我们看看这个代码在哪里用的。

在 broker 模块的 BrokerController 类中，有个 registerProcessor 方法，会将 SEND_MESSAGE Code 和一个 SendMessageProcessor 对象绑定。

NettyRemotingServer是处理Request 的类，ServerBootstrap 会在 pipeline 中添加一个 NettyServerHandler处理器，这个处理器的channelRead0方法会调用 NettyRemotingServer的父类processMessageReceived 方法。

从processorTable 里，根据 Cmd Code，也就是 SEND_MESSAGE 获取对应的 Processor

一部分是处理数据的对象，一部分是这个对象所对应的线程池。用于异步处理逻辑，防止阻塞 Netty IO线程。

doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);

前后都是执行一些钩子，例如 ACL

RocketMQ会有一个 BrokerController 类，会注册 Code 和 Processor 的绑定关系，BrokerController 也会把这些绑定，注册到 Netty Server 中，当 Netty Server 从 Socket 收到 Cmd 对象，根据 Cmd 对象的 Code，就可以找到对应 Processor 类，对数据进行处理。

中间是处理 Request请求的。这个 processRequest 方法，有很多的实现，SendMessageProcessor的sendMessage 是处理消息的主要逻辑。

消息存储引擎，这里我们看DefaultMessageStore的putMessage 实现。

putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);

由于RocketMQ写数据是PageCache里面写的，因此，如果写的慢，就是 PageCache 忙，这里忙的标准是，如果锁文件的时间，超过了 1 秒，那就是忙。

if (this.isOSPageCacheBusy()) {return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
}

最后调用 PutMessageResult result = this.commitLog.putMessage(msg) 写数据。如果耗时超过 500 毫秒，就会打印日志。这样我们排查问题的时候，可以看看 storeStats 的日志。

result = mappedFile.appendMessage(msg, this.appendMessageCallback)

写完之后，释放锁，如果超过 500 毫秒，打印 cost time 日志。

处理刷盘和slave 同步，这里看刷盘策略和同步策略，是 SYNC 还是 ASYNC。经过我的测试，同步刷盘和异步刷盘的性能差距是 10 倍。

而 Slave 的数据同步，如果用 SYNC 模式，tps 最高也就 2000 多一丢度，为什么？内网，两台机器 ping 一下都要 0.2 毫秒，一秒最多 5000 次，再加上处理逻辑， 2000 已经到顶了，网络成了瓶颈。

我们看看 mappedFile.appendMessage 方法的实现。一路追踪，有个关键逻辑，在 appendMessagesInner 里：

int currentPos = this.wrotePosition.get();
if (currentPos < this.fileSize) {ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();byteBuffer.position(currentPos);AppendMessageResult result = null;if (messageExt instanceof MessageExtBrokerInner) {result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);} else if (messageExt instanceof MessageExtBatch) {result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);} else {return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);}this.wrotePosition.addAndGet(result.getWroteBytes());this.storeTimestamp = result.getStoreTimestamp();return result;
}