RocketMQ源码学习

叁歲伎倆 2023-05-30 10:51 166阅读 0赞

# RocketMQ源码学习 #

### 文章目录 ###

*  RocketMQ源码学习
 *   *  Producer 是怎么将消息发送至 Broker 的？
     *   *  同步发送
         *  异步发送
         *  队列选择器
         *  事务消息
         *   *  原理
     *  Broker 是怎么处理客户端发送的消息？
     *   *  NettyRequestProcessor
         *  发送消息
         *  事务消息
         *   *  处理发送消息请求
             *  处理发送结束事务请求
             *  定时任务回查逻辑
             *   *  Broker端
                 *  客户端
     *  客户端是怎样与服务端交互的？
     *  客户端怎么获取 Broker 信息？
     *   *  客户端
         *  NameServer端
     *  NameServer 是怎么管理 Broker 信息的？
     *  客户端怎么获取队列信息？
     *  Broker 是怎么存储消息的？
     *  RocketMQ 的消息存储架构是怎样的？

## Producer 是怎么将消息发送至 Broker 的？ ##

### 同步发送 ###

[RocketMQ 源码学习笔记 Producer 是怎么将消息发送至 Broker 的？][RocketMQ _ Producer _ Broker]

### 异步发送 ###

*  `DefaultMQProducer#send(Message, SendCallback)`
 *  `DefaultMQProducerImpl#send(Message, SendCallback)`
 *  `DefaultMQProducerImpl#send(Message, SendCallback, long)`
 *  `DefaultMQProducerImpl#sendDefaultImpl(Message, CommunicationMode, SendCallback, long)`
 *  `DefaultMQProducerImpl#sendKernelImpl(Message, MessageQueue, CommuicationMode, SendCallback, TopicPublishInfo, long)`
 *  `MQClientAPIImpl#sendMessage(String, String, Message, SendMessageRequestHeader, long, CommunicationMode, SendCallback, TopicPublishInfo, MQClientInstance, int, SendMessageContext, DefaultMQProducerImpl)`
 *  `MQClientAPIImpl#sendMessageAsync(String, String, Message, long, RemotingCommand, SendCallback, TopicPublishInfo, MQClientInstance, int, AtomicInteger, SendMessageContext, DefaultMQProducerImpl)`
 *  `NettyRemotingClient#invokeAsync(String, RemotingCommand, long, InvokeCallback)`
 *  `NettyRemotingAbstract#invokeAsyncImpl(Channel, RemotingCommand, long, InvokeCallback)`
    
     *  `this.responseTable.put(opaque, responseFuture);`
     *  将`responseFuture`放入`map`中，等待处理（在`netty`接收中服务端返回的响应）
     *  在该方法中处理：`NettyRemotingAbstract#processResponseCommand`

### 队列选择器 ###

*  `DefaultMQProducer#send(Message, MessageQueueSelector, Object, SendCallback)`
 *  `DefaultMQProducerImpl#send(Message, MessageQueueSelector, Object, SendCallback)`
 *  `DefaultMQProducerImpl#send(Message, MessageQueueSelector, Object, SendCallback, long)`
 *  `DefaultMQProducerImpl#sendSelectImpl(Message, MessageQueueSelector, Object, CommunicationMode, SendCallback, timeout)`
    
     *  `mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));`
        
         *  `ClientConfig#queueWithNamespace(MessageQueue)`
         *  `RocketMQ`提供的`selector`
            
             *  `org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash`
             *  `org.apache.rocketmq.client.producer.selector.SelectMessageQueueByMachineRoom`
             *  `org.apache.rocketmq.client.producer.selector.SelectMessageQueueByRandom`
 *  `DefaultMQProducerImpl#sendKernelImpl(Message, MessageQueue, CommuicationMode, SendCallback, TopicPublishInfo, long)`
 *  接下来的调用见异步发送

### 事务消息 ###

#### 原理 ####

![极客时间-消息队列高手课-04|如何利用事务消息实现分布式事务？][-_-04]

*  `TransactionListener`：`TransactionMQProducer#setTransactionListener`
    
     *  执行本地事务：`TransactionListener#executeLocalTransaction`
     *  检查本地事务：`TransactionListener#checkLocalTransaction`
     *  执行状态：提交消息：`COMMIT_MESSAGE`；回滚消息：`ROLLBACK_MESSAGE`；未知状态：`UNKNOW`
 *  `TransactionMQProducer#sendMessageInTransaction(Message, Object)`
 *  `DefaultMQProducerImpl#sendMessageInTransaction(Message, LocalTransactionExecuer, Object)`
    
     *  `DefaultMQProducerImpl#send(Message)`
        
         *  `DefaultMQProducerImpl#send(Message)`
         *  接下来的调用见同步发送
     *  `TransactionListener#executeLocalTransaction(Message, Object)`
     *  `DefaultMQProducerImpl#endTransaction(SendResult, LocalTransactionState, Throwable)`
        
         *  `MQClientAPIImpl#endTransactionOneway(String, EndTransactionREquestHeader, String, long)`
         *  `RemotingClient#invokeOneway(String, RemotingCommand, long)`
         *  发送消息给Broker: `NettyRemotingAbstract#invokeOnewayImpl(Channel, RemotingCommand, long)`

## Broker 是怎么处理客户端发送的消息？ ##

`Broker`接收客户端发过来的消息是从`NettyRemotingAbstract#processMessageReceived(ChannelHandlerContext, RemotingCommand)`开始的。在该方法中，通过`RemotingCommand#getType()`来判断是进入请求命令处理分支，还是响应命令处理分支。

因为是接收客户端的请求命令，所以这里进入请求命令分支：`NettyRemotingAbstract#processRequestCommand(ChannelHandlerContext, RemotingCommand)`

在该方法中，通过`this.processorTable.get(cmd.getCode())`获取对应的`NettyRequestProcessor`处理器。这里使用的是状态模式，通过不同的`code`值来执行不同的逻辑。只不过这种模式比较巧妙，预先定义处理逻辑，将状态和处理逻辑作为键值对存入`map`中，通过`map#get(status)`这样的操作来获取状态处理逻辑。使用这种方式，要注意提供默认逻辑，当`status`找不到对应的处理逻辑时，默认执行该逻辑。当然，在`RocketMQ`中已经提供了默认逻辑`null == matched ? this.defaultRequestProcessor : matched`。

然后构建一个`RequestTask`对象，并执行它。

### NettyRequestProcessor ###

package org.apache.rocketmq.remoting.netty;
    
    import io.netty.channel.ChannelHandlerContext;
    import org.apache.rocketmq.remoting.protocol.RemotingCommand;
    
    /** * Common remoting command processor */
    public interface NettyRequestProcessor { 
        RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception;
        
        boolean rejectRequest();
    }

服务端注册处理器的地方在`BrokerController#registerProcessor()`

### 发送消息 ###

发送消息的处理器为`SendMessageProcessor`

*  `SendMessageProcessor#processRequest(ChannelHandlerContext, RemotingCommand)`
    
     *  `AbstractSendMessageProcessor#parseRequestHeader(RemotingCommand)`
     *  `SendMessageProcessor#sendMessage(ChannelHandlerContext, RemotingCommand, SendMessageContext, SendMessageRequestHeader)`
        
         *  `MessageStore#putMessage(MessageExtBrokerInner)`
            
             *  `CommitLog#putMessage(MessageExtBrokerInner)`
                
                 *  `MappedFile#appendMessage(MessageExtBrokerInner, AppendMessageCallback)`
                    
                     *  `MappedFile#appendMessagesInner(MessageExt, AppendMessageCallback)`
                        
                         *  在该回调方法中 append 消息`DefaultAppendMessageCallback#doAppend(long, ByteBuffer, int, MessageExtBrokerInner)`
         *  `SendMessageProcessor#handlePutMessageResult(PutMessageResult, RemotingCommand, RomotingCommand, MessageExt, SendMessageResponseHeader, SendMessageContext, ChannelHadnlerContext, int)`

### 事务消息 ###

事务消息在服务端有：处理发送消息请求、处理发送结束事务请求两次请求处理，还有一个定时任务回查逻辑。

#### 处理发送消息请求 ####

发送消息请求与发送消息处理方式是一样的，不过在`SendMessageProcessor#sendMessage`中会判断是否事务消息并进行处理。

// 事务标识
    String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    if (traFlag != null && Boolean.parseBoolean(traFlag)) {   // 判断事务标识
        if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) { 
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark(
                "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                    + "] sending transaction message is forbidden");
            return response;
        }
        // 保存事务半消息
        putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
    } else { 
        putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
    }

*  `SendMessageProcessor#sendMessage(ChannelHandlerContext, RemotingCommand, SendMessageContext, SendMessageRequestHeader)`
 *  `TransactionalMessageServiceImpl#prepareMessage(MessageExtBrokerInner)`
 *  `TransactionalMessageBridge#putHalfMessage(MessageExtBrokerInner)`
    
     *  保存半消息：`store.putMessage(parseHalfMessageInner(messageInner))`

#### 处理发送结束事务请求 ####

发送结束事务请求的处理器是`EndTransactionProcessor`

*  `EndTransactionProcessor#processRequest(ChannelHandlerContext, RemotingCommand)`

#### 定时任务回查逻辑 ####

定时任务回查分为服务端与客户端两块逻辑

##### Broker端 #####

*  初始化定时任务检测对象：`BrokerController#initialTransaction()`
 *  启动定时任务：`BrokerController#start()`
 *  `TransactionalMessageCheckService#onWaitEnd()`
 *  `TransactionalMessageServiceImpl#check(long, int, AbstractTransactionalMessageCheckListener)`
 *  `AbstractTransactionalMessageCheckListener#resolveHalfMsg(MessageExt)`
 *  `AbstractTransactionalMessageCheckListener#sendCheckMessage(MessageExt)`
 *  发送检测消息：`Broker2Client#checkProducerTransactionState(String, Channel, CheckTransactionStateRequestHeader, MessageExt)`

##### 客户端 #####

*  客户端通用处理请求方法：`ClientRemotingProcessor#processRequest(ChannelHandlerContext, RemotingCommand)`
 *  `ClientRemotingProcessor#checkTransactionState(ChannelHandlerContext, RemotingCommand)`
 *  `DefaultMQProducerImpl#checkTransactionState(String, MessageExt, CheckTransactionStateRequestHeader)`
 *  `MQClientAPIImpl#endTransactionOneway(String, EndTransactionRequestHeader, String, long)`
 *  接下来的调用见客户端事务消息

## 客户端是怎样与服务端交互的？ ##

以从`NameServer`获取路由信息为例：

*  `MQClientAPIImpl#getTopicRouteInfoFromNameServer(String, long, boolean)`

public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
        boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException { 
        // 创建一个实现了 CommandCustomHeader 接口的对象
        GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
        requestHeader.setTopic(topic);
        // 通过请求码和 header 创建一个 RemotingCommand 对象
        // RequestCode 中有不同的请求码，通过请求码来确定请求类型，并进行相应处理
        // 该创建方式为通用创建请求对象方式
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader);
        
        // 同步调用获取结果
        // 内部调用处理方式和 Broker 是怎么处理客户端发送的消息一节原理相似
        RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) { 
            case ResponseCode.TOPIC_NOT_EXIST: { 
                if (allowTopicNotExist && !topic.equals(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC)) { 
                    log.warn("get Topic [{}] RouteInfoFromNameServer is not exist value", topic);
                }
    
                break;
            }
            case ResponseCode.SUCCESS: { 
                byte[] body = response.getBody();
                if (body != null) { 
                    // 返回解码反序列化后的响应体
                    return TopicRouteData.decode(body, TopicRouteData.class);
                }
            }
            default:
                break;
        }
    
        throw new MQClientException(response.getCode(), response.getRemark());
    }

由上面一例，可知：

1.  请求是基于`RemotingClient`对象的，`RemotingClient`是一个接口，所以可以通过选择不同的实现类，选择服务器支持的协议进行交互。
2.  默认使用`NettyRemotingClient`对象进行交互。基于`netty`通讯框架。
3.  请求对象与响应对象同为`RemotingCommand`，简化了通讯框架序列化与反序列化的代码。
4.  `RequestCode`中有不同的请求码，通过请求码来确定请求类型，并进行相应处理
5.  `ResponseCode`中有不同的响应码，通过响应码来确定响应类型，并进行相应处理
6.  通过实现`CommandCustomHeader`接口，来实现`header`的通用化处理。
    
     *  使用此方式需要注意序列化与反序列化时，对象类型及对象中字段的值是否正确处理

## 客户端怎么获取 Broker 信息？ ##

以获取`Broker`集群信息为例

### 客户端 ###

`MQClientAPIImpl`是客户端API实现类，通过该类可以了解客户端提供了哪些API接口供上层调用。

*  `MQClientAPIImpl#getBrokerClusterInfo(long)`
    
     *  `RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);`
     *  `RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);`

### NameServer端 ###

*  `DefaultRequestProcessor#processRequest(ChannelHandlerContext, RemotingCommand)`
 *  `DefaultRequestProcessor#getBrokerClusterInfo(ChannelHandlerContext, RemotingCommand)`
 *  `RouteInfoManager#getAllClusterInfo()`

public byte[] getAllClusterInfo() { 
        // 创建集群信息类
        ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
        // 设置 broker 名与 broker 地址的映射对象
        clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
        // 设置 集群名与 broker 集合的映射对象
        clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
        // 返回编码后的对象
        return clusterInfoSerializeWrapper.encode();
    }

## NameServer 是怎么管理 Broker 信息的？ ##

## 客户端怎么获取队列信息？ ##

## Broker 是怎么存储消息的？ ##

## RocketMQ 的消息存储架构是怎样的？ ##

[RocketMQ _ Producer _ Broker]: https://app.yinxiang.com/shard/s67/nl/15352878/4b5a6f1d-a581-4fbe-8ecd-61f4594e8db5/
[-_-04]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9zd2l0Y2gtb3BlbndyaXRlLm9zcy1jbi1oYW5nemhvdS5hbGl5dW5jcy5jb20vaW1hZ2Utcm9ja2V0LW1xLXRyYW5zYWN0aW9uLXRoZW9yeS5qcGc?x-oss-process=image/format,png