RocketMQ源码12-broker 消息投递流程(处理PULL_MESSAGE请求)

谁践踏了优雅 2024-03-24 06:16 81阅读 0赞

`RocketMq`消息处理整个流程如下：

![format_png][]

1.  消息接收：消息接收是指接收`producer`的消息，处理类是`SendMessageProcessor`，将消息写入到`commigLog`文件后，接收流程处理完毕；
2.  消息分发：`broker`处理消息分发的类是`ReputMessageService`，它会启动一个线程，不断地将`commitLong`分到到对应的`consumerQueue`，这一步操作会写两个文件：`consumerQueue`与`indexFile`，写入后，消息分发流程处理 完毕；
3.  消息投递：消息投递是指将消息发往`consumer`的流程，`consumer`会发起获取消息的请求，`broker`收到请求后，调用`PullMessageProcessor`类处理，从`consumerQueue`文件获取消息，返回给`consumer`后，投递流程处理完毕。

以上就是`rocketMq`处理消息的流程了，接下来我们就从源码来分析消息投递的实现。

## 1. 处理`PULL_MESSAGE`请求 ##

与`producer`不同，`consumer`从`broker`拉取消息时，发送的请求`code`为`PULL_MESSAGE`，`processor`为`PullMessageProcessor`，我们直接进入它的`processRequest`方法：

@Override
    public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingCommand request)
            throws RemotingCommandException {
        // 调用方法
        return this.processRequest(ctx.channel(), request, true);
    }
    复制代码

这个方法就只是调用了一个重载方法，多出来的参数`true`表示允许`broker`挂起请求，我们继续，

/**
     * 继续处理
     */
    private RemotingCommand processRequest(final Channel channel, RemotingCommand request, 
            boolean brokerAllowSuspend)throws RemotingCommandException {
    
        RemotingCommand response = RemotingCommand
            .createResponseCommand(PullMessageResponseHeader.class);
        final PullMessageResponseHeader responseHeader 
            = (PullMessageResponseHeader) response.readCustomHeader();
        final PullMessageRequestHeader requestHeader = (PullMessageRequestHeader) 
            request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
    
        response.setOpaque(request.getOpaque());
    
        // 省略权限校验流程
        // 1. rocketMq 可以设置校验信息，以阻挡非法客户端的连接
        // 2. 同时，对topic可以设置DENY(拒绝)、ANY(PUB 或者 SUB 权限)、PUB(发送权限)、SUB(订阅权限)等权限，
        //    可以细粒度控制客户端对topic的操作内容
        ...
    
        // 获取订阅组
        SubscriptionGroupConfig subscriptionGroupConfig =
            this.brokerController.getSubscriptionGroupManager()
            .findSubscriptionGroupConfig(requestHeader.getConsumerGroup());
        ...
    
        // 获取订阅主题
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager()
            .selectTopicConfig(requestHeader.getTopic());
        ...
    
        // 处理filter
        // consumer在订阅消息时，可以对订阅的消息进行过滤，过滤方法有两种：tag与sql92
        // 这里我们重点关注拉取消息的流程，具体的过滤细节后面再分析
        ...
    
        // 获取消息
        // 1. 根据 topic 与 queueId 获取 ConsumerQueue 文件
        // 2. 根据 ConsumerQueue 文件的信息，从 CommitLog 中获取消息内容
        final GetMessageResult getMessageResult = this.brokerController.getMessageStore().getMessage(
            requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), 
            requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);
        if (getMessageResult != null) {
            // 省略一大堆的校验过程
            ...
    
            switch (response.getCode()) {
                // 表示消息可以处理，这里会把消息内容写入到 response 中
                case ResponseCode.SUCCESS:
                    ...
                    // 处理消息消息内容，就是把消息从 getMessageResult 读出来，放到 response 中
                    if (this.brokerController.getBrokerConfig().isTransferMsgByHeap()) {
                        final long beginTimeMills = this.brokerController.getMessageStore().now();
                        // 将消息内容转为byte数组
                        final byte[] r = this.readGetMessageResult(getMessageResult, 
                            requestHeader.getConsumerGroup(), requestHeader.getTopic(), 
                            requestHeader.getQueueId());
                        ...
                        response.setBody(r);
                    } else {

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/99ebbc94bb7947b6bea5224b50a0956d.png