RocketMQ源码（一）RocketMQ消息生产及消费通信链路源码分析
RocketMQ的核心架构主要分为Broker、Producer、Consumer，通过阅读源码看到他们之间是通过Netty来通信的
，具体来说Broker端是Netty服务器用来负责与客户端的连接请求处理，而Producer/Consumer端是Netty客户端用来负责与Netty服务器的通信及请求响应处理。
Tip：我本人在多年的开发经验中常用的MQ中间件如Kafka、RocketMQ都实战使用的，简单的使用示例可以参考我Git。
// Git代码
https://gitee.com/yeeevip/yeee-memo/tree/master/middle-ware/mq
https://github.com/yeeevip/yeee-memo/tree/master/middle-ware/mq
1 Netty客户端/服务端运行
1.1 Broker端Netty服务器
- BrokerController引导启动Netty服务器
创建NettyRemotingServer对象并调用start()方法来启动Netty服务器
public class BrokerController {
protected void initializeRemotingServer() {
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
...
}
protected void startBasicService() throws Exception {
...
if (this.remotingServer != null) {
this.remotingServer.start();
}
...
}
}
- Netty服务端启动
这里就是我们熟悉的Netty的ServerBootstrap服务端引导类，通过设置EventLoopGroup然后绑定端口接着添加一系列的ChannelHandler启动服务器；
对于RocketMQ来说主要的Handler是NettyServerHandler这个处理类，它主要负责接收客户端请求并进行处理。
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
// sharable handlers
private NettyServerHandler serverHandler;
@Override
public void start() {
...
// 用于处理客户端请求命令
serverHandler = new NettyServerHandler();
...
serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
...
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline()
...
.addLast(defaultEventExecutorGroup, ..., serverHandler);
}
});
try {
ChannelFuture sync = serverBootstrap.bind().sync();
...
} catch (Exception e) {
...
}
...
}
}
1.2 Producer/Consumer端Netty客户端
Producer和Consumer的启动最终都会调用mQClientFactory.start()来创建Netty客户端
// Producer
public class DefaultMQProducerImpl implements MQProducerInner {
private MQClientInstance mQClientFactory;
public void start(final boolean startFactory) throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
...
// 这里创建Netty客户端
mQClientFactory.start();
...
case RUNNING:
...
default:
break;
}
...
}
}
// Consumer
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
private MQClientInstance mQClientFactory;
public synchronized void start() throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
...
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
...
// 这里创建Netty客户端
mQClientFactory.start();
...
case RUNNING:
...
default:
break;
}
...
}
}
- 创建Netty客户端对象
执行mQClientFactory.start() --> this.mQClientAPIImpl.start() --> this.remotingClient.start()，最终NettyRemotingClient对象调用start()方法来创建Netty客户端；
Netty客户端由Bootstrap引导程序创建，之后请求/响应通过Netty客户端处理。
public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient {
@Override
public void start() {
this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class)
...
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
...
ch.pipeline().addLast(..., new NettyClientHandler());
}
});
...
}
}
2 消息的生产及保存
2.1 Producer生产消息到Broker
我们调用producer.send发送消息时，程序会使用RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE)把消息封装为自定义的通信协议RemotingCommand，
之后NettyRemotingClient会找到Broker地址并建立连接生成Channel对象调用writeAndFlush方法将请求（RemotingCommand）发送到Netty服务器
- 使用producer.send发送同步消息
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
producer.start();
Message msg = new Message(TOPIC /* Topic */,
TAG /* Tag */,
("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
SendResult sendResult = producer.send(msg);
}
}
- 封装为RemotingCommand消息协议进一步执行发送逻辑
public class MQClientAPIImpl implements NameServerUpdateCallback {
public SendResult sendMessage(final Message msg, final SendMessageRequestHeader requestHeader, ...) {
// 构建请求命令 RequestCode.SEND_MESSAGE
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
request.setBody(msg.getBody());
switch (communicationMode) {
...
case SYNC:
return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
...
}
return null;
}
}
- 调用与服务端建立的Channel的writeAndFlush方法将请求RemotingCommand发送到Broker
public abstract class NettyRemotingAbstract {
public RemotingCommand invokeSyncImpl(Channel channel, RemotingCommand request) {
...
try {
...
// 向服务端Broker通道发送消息
channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
...
});
RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
...
return responseCommand;
}
...
}
}
2.2 Broker接收消息并保存
在开始介绍说Broker中NettyServer的启动时会添加NettyServerHandler一个处理器，这个handler负责处理client发过来的请求指令。
上面当客户端Producer发送RemotingCommand(RequestCode.SEND_MESSAGE)这个指令的请求时，Broker收到请求后通过RequestCode
找到对应的SendMessageProcessor处理器执行processRequest方法去处理消息接收后的逻辑。
- SendMessageProcessor处理器来接收消息调用MessageStore保存消息
public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
...
response = this.sendMessage(ctx, request, sendMessageContext, requestHeader, mappingContext,
(ctx12, response12) -> executeSendMessageHookAfter(response12, ctx12));
...
}
public RemotingCommand sendMessage(...， final RemotingCommand request, ...) throws RemotingCommandException {
...
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
...
putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
...
}
}
- MessageStore保存消息到CommitLog中
public class DefaultMessageStore implements MessageStore {
@Override
public PutMessageResult putMessages(MessageExtBatch messageExtBatch) {
return waitForPutResult(asyncPutMessages(messageExtBatch));
}
@Override
public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch) {
...
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessages(messageExtBatch);
...
}
}
public class CommitLog implements Swappable {
public CompletableFuture<PutMessageResult> asyncPutMessages(final MessageExtBatch messageExtBatch) {
...
result = mappedFile.appendMessages(messageExtBatch, this.appendMessageCallback, putMessageContext);
...
}
}
3 消息的拉取及消费
RocketMQ的核心架构主要分为Broker、Producer、Consumer，通过阅读源码看到他们之间是通过Netty来通信的
，具体来说Broker端是Netty服务器用来负责与客户端的连接请求处理，而Producer/Consumer端是Netty客户端用来负责与Netty服务器的通信及请求响应处理。
3.1 Consumer向Broker发送拉取请求
创建Consumer实例，订阅Topic并注册MessageListener后调用start方法启动程序；
接着开启一个PullMessageService任务去向Broker发送消息拉取请求，通过DefaultMQPushConsumerImpl.pullMessage方法设置请求回调逻辑(如：获取到消息则使用MessageListener去消费)，
接着继续执行MQClientAPIImpl().pullMessage将请求信息封装为RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE)，最终由NettyRemotingClient发送请求到Broker。
- 创建DefaultMQPushConsumer调用start()启动消费者程序，PullMessageService任务开启
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
consumer.subscribe(TOPIC, "*");
consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
}
}
public class MQClientInstance {
public void start() throws MQClientException {
...
// Start pull service
this.pullMessageService.start();
...
}
}
- 这里会设置回调逻辑等并继续调用底层实现发起请求，开始注册的MessageListener会在这里调用的
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
public void pullMessage(final PullRequest pullRequest) {
...
PullCallback pullCallback = new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
if (pullResult != null) {
switch (pullResult.getPullStatus()) {
// 获取到消息
case FOUND:
...
// 这个里面会执行注册的MessageListener
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(pullResult.getMsgFoundList(), ...);
...
// 继续拉取消息
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
...
...
}
}
}
...
};
...
try {
this.pullAPIWrapper.pullKernelImpl(pullRequest.getMessageQueue(), ..., pullRequest.getNextOffset(), ..., pullCallback);
} catch (Exception e) {
...
}
}
}
上面执行到pullAPIWrapper.pullKernelImpl会调用MQClientAPIImpl().pullMessage来封装请求报文最终交由NettyRemotingClient.invokeSyncImpl真正发出请求。
- 封装请求报文RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE)
public class MQClientAPIImpl implements NameServerUpdateCallback {
public PullResult pullMessage(final String addr, final PullMessageRequestHeader requestHeader, ..., final PullCallback pullCallback) {
...
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader);
...
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
...
}
}
- RemotingClient调用channel.writeAndFlush(request)发出拉取请求
public abstract class NettyRemotingAbstract {
public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis) {
...
try {
...
channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
...
});
...
}
...
}
}
3.2 Broker接收拉取请求匹配返回消息
同样Broker中NettyServerHandler收到RemotingCommand(RequestCode.PULL_MESSAGE)这个指令找到PullMessageProcessor调用processRequest方法去处理。
- PullMessageProcessor.processRequest处理时，调用messageStore.getMessageAsync去队列里查找消息，之后写回客户端
public class PullMessageProcessor implements NettyRequestProcessor {
private RemotingCommand processRequest(final Channel channel, RemotingCommand request, ...) {
...
messageStore.getMessageAsync(group, topic, queueId, requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter)
.thenApply(result -> {
...
})
// 写回客户端
.thenAccept(result -> NettyRemotingAbstract.writeResponse(channel, request, result));
...
}
}
- PullMessageProcessor.processRequest处理时，调用messageStore.getMessageAsync去队列里查找消息，之后写回客户端
public class DefaultMessageStore implements MessageStore {
@Override
public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset, ...) {
...
GetMessageResult getResult = new GetMessageResult();
...
SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);
...
getResult.addMessage(selectResult, cqUnit.getQueueOffset(), cqUnit.getBatchNum());
...
return getResult;
}
}
- 从commitLog中获取消息
public class CommitLog implements Swappable {
public SelectMappedBufferResult getMessage(final long offset, final int size) {
int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
if (mappedFile != null) {
int pos = (int) (offset % mappedFileSize);
SelectMappedBufferResult selectMappedBufferResult = mappedFile.selectMappedBuffer(pos, size);
if (null != selectMappedBufferResult) {
selectMappedBufferResult.setInCache(coldDataCheckService.isDataInPageCache(offset));
return selectMappedBufferResult;
}
}
return null;
}
}
最后
至此我们把RocketMQ中Broker与生产者/消费者基于Netty简单的通信调用链路讲完了，大家有什么问题可以下面留言哦，一起学习进步啊。
Tip：我本人在多年的开发经验中常用的MQ中间件如Kafka、RocketMQ都实战使用的，简单的使用示例可以参考我Git。
// Git代码
https://gitee.com/yeeevip/yeee-memo/tree/master/middle-ware/mq
https://github.com/yeeevip/yeee-memo/tree/master/middle-ware/mq
