投稿
  1. 首页
  2. python

rocketmq源码分析

code • 2022-4-27 • python • 阅读 45

rocketmq源码分析,第1张

文章目录
  • 前言
  • 一、NameServer路由中心
  • 二、消息存储
    • 1.消息发送存储流程
    • 2.存储文件
  • 三、消息消费
    • 1、消息者启动流程
    • 2、消息拉取
      • 1、PullMessageService实现机制
      • 2、ProcessQueue实现机制
      • 3、消息拉取基本流程
          • 3.1客户端封装消息拉取请求
          • 3.2消息服务端Broker组装信息
          • 3.3消息拉取客户端处理消息
          • 3.4消息拉取长轮询机制分析
          • 3.5PullRequestHoldService
          • 3.6DefaultMessageStore$ReputMessageService详解
  • 总结

前言

疫情期间，学习研究了一下《RocketMQ技术内幕》这本书，记录一下自己的学习心得

一、NameServer路由中心

待续

二、消息存储 1.消息发送存储流程

代码如下（示例）：

待续
2.存储文件

待续

三、消息消费 1、消息者启动流程

跟踪DefaultMQPushConsumer的启动流程，start()方法

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TopicTest", "*");
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}

DefaultMQPushConsumerImpl#start

1、构建主题订阅信息SubscriptionData并加入RebalanceImpl的订阅消息中。
2、初始化MQClientInstance、RebalanceImpl（消息重新负载实现类）等
3、初始化消息进度。集群模式进度保存在Broker上，广播模式季度存储在消费端。
4、根据是否顺序消费，创建消费端消费线程服务。ConsumeMessageService主要负责消息消费，内部维护一个线程池
5、向MQClientInstance注册消息消费者，并启动MQClientInstance，在一个JVM中的所有消费者、生产者持有同一个MQClientInstance、MQClientInstance只会启动一次

/**
     * 消费启动流程
     *
     * @throws MQClientException
     */
    public synchronized void start() throws MQClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
                    this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
                this.serviceState = ServiceState.START_FAILED;

                this.checkConfig();
                // 1、构建主题订阅信息SubscriptionData并加入RebalanceImpl的订阅消息中。
                this.copySubscription();

                // 2、初始化MQClientInstance、RebalanceImpl（消息重新负载实现类）等
                if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                    this.defaultMQPushConsumer.changeInstanceNameToPID();
                }

                this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);

                this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);

                this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

                // 3、初始化消息进度。集群模式进度保存在Broker上，广播模式季度存储在消费端。
                if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                } else {
                    switch (this.defaultMQPushConsumer.getMessageModel()) {
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }
                    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                }
                this.offsetStore.load();

                // 4、根据是否顺序消费，创建消费端消费线程服务。ConsumeMessageService主要负责消息消费，内部维护一个线程池
                if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                    this.consumeOrderly = true;
                    this.consumeMessageService =
                        new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                    this.consumeOrderly = false;
                    this.consumeMessageService =
                        new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                }

                this.consumeMessageService.start();

                // 5、向MQClientInstance注册消息消费者，并启动MQClientInstance，
                // 在一个JVM中的所有消费者、生产者持有同一个MQClientInstance、MQClientInstance只会启动一次
                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                if (!registerOK) {
                    this.serviceState = ServiceState.CREATE_JUST;
                    this.consumeMessageService.shutdown();
                    throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                        null);
                }

                mQClientFactory.start();
                log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
                this.serviceState = ServiceState.RUNNING;
                break;
            case RUNNING:
            case START_FAILED:
            case SHUTDOWN_ALREADY:
                throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
                    + this.serviceState
                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                    null);
            default:
                break;
        }

        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        this.mQClientFactory.checkClientInBroker();
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        this.mQClientFactory.rebalanceImmediately();
    }
2、消息拉取 1、PullMessageService实现机制

承接上文，当启动MQClientInstance时，触发MQClientInstance.start()方法，其中就触发了消息拉取Runnable任务–this.pullMessageService.start();

轮询从pullRequestQueue阻塞队列获取PullRequest，然后执行pullMessage(pullRequest)拉取消息，进行消费

PullRequest核心属性

问题一：PullRequest是什么时候添加的？

问题二：PullRequest是什么时候创建的？

1、RocketMQ根据PullRequest拉取任务执行完一次消息拉取任务后，又将PullRequest对象放入到pullRequestQueue
2、RebalanceImpl中创建，是PullRequest对象真正创建的地方！！

2、ProcessQueue实现机制

ProcessQueue是MessageQueue在消费端的重新、快照。消息服务器默认每次拉取32条消息，按消息的队列偏移量顺序存在在ProcessQueue中，PullMessageService然后将消息提交到消费者消费线程池，消息成功消费后从ProcessQueue中移除

ProcessQueue核心属性

3、消息拉取基本流程 3.1客户端封装消息拉取请求

DefaultMQPushConsumerImpl#pullMessage

消息拉取的基本流程：
1、消息拉取客户端–消息拉取请求封装
2、消息服务器-----查找并返回信息
3、消息拉取客户端–处理返回信息

1、如果当前消费者被挂起，则将拉取任务延迟1s再次放入到PullMessageService的拉取任务队列中【pullRequestQueue】，结束本次消息拉取

2、进行消息拉取流控

3、拉取主题订阅信息，如果为空，结束本次消息拉取，下次任务延迟3s

4、构建消息拉取系统标记【PullSysFlag】

PullSysFlag核心属性

5、调用PullAPIWrapper.pullKernelImpl方法后与服务端交互。

pullKernelImpl方法核心参数

netty发送请求到Broker消息服务端

消息拉取命令Code：RequestCode.PULL_MESSAGE

3.2消息服务端Broker组装信息

根据RequestCode.PULL_MESSAGE找到对应request请求处理类，PullMessageProcessor#processRequest

1、根据订阅信息，构建消息过滤器。
2、MessageStore.getMessage 查找消息
3、根据主题名称与队列编号获取消息消费队列
4、消息偏移量异常情况校对下一次拉取偏移量
5、如果待拉去偏移量offset大于minOffset并且小于maxOffset，从当前offset处尝试拉取32条信息。
6、根据getMessageResult填充responseHeader的nextBeginOffset、minOffset、maxOffset
7、根据主从同步延迟，如果从节点数据包含下一次拉取的偏移量，设置下一次拉取任务的brokerId
8、根据getMessageResult.getStatus()编码转换关系
9、如果commitlog标记可用并且当前节点为主节点，则更新消息消费进度





服务端消息拉取处理完毕，将返回结果到拉取消息调用方。在调用方，需要重点关注PULL_RETRY_IMMEDIATELY、PULL_OFFSET_MOVED、PULL_NOT_FOUND等情况下如何矫正拉取偏移量

3.3消息拉取客户端处理消息

NettyRemotingClient在收到服务端响应结构后会回调PullCallback的onSuccess或onException
PullCallback对象在DefaultMQPushConsumerImpl#pullMessage中创建

消息拉取客户端调用入口（处理服务端返回的response）
MQClientAPIImpl#pullMessageAsync

1、响应结果解码成PullResultExt对象

private PullResult processPullResponse(
        final RemotingCommand response) throws MQBrokerException, RemotingCommandException {
        PullStatus pullStatus = PullStatus.NO_NEW_MSG;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS:
                pullStatus = PullStatus.FOUND;
                break;
            case ResponseCode.PULL_NOT_FOUND:
                pullStatus = PullStatus.NO_NEW_MSG;
                break;
            case ResponseCode.PULL_RETRY_IMMEDIATELY:
                pullStatus = PullStatus.NO_MATCHED_MSG;
                break;
            case ResponseCode.PULL_OFFSET_MOVED:
                pullStatus = PullStatus.OFFSET_ILLEGAL;
                break;

            default:
                throw new MQBrokerException(response.getCode(), response.getRemark());
        }

        PullMessageResponseHeader responseHeader =
            (PullMessageResponseHeader) response.decodeCommandCustomHeader(PullMessageResponseHeader.class);

        // 响应结果解码成PullResultExt对象
        return new PullResultExt(pullStatus, responseHeader.getNextBeginOffset(), responseHeader.getMinOffset(),
            responseHeader.getMaxOffset(), null, responseHeader.getSuggestWhichBrokerId(), response.getBody());
    }

2、调用pullAPIWrapper.processPullResult将消息字节数组解码成消息列表，填充PullResult的属性msgFoundList，并对消息进行消息过滤(TAG)模式

3、更新PullRequest的下一次拉取偏移量，如果msgFoundList为空，则立即将PullRequest放入到PullMessageService的pullRequestQueue，以便PullMessageService能及时环形并再次执行消息拉取

将拉取到的消息存入ProcessQueue 存入消费端本地缓存
然后将拉取到的消息提交到ConsumeMessageService中供消费者消费 线程池异步处理
如果pullInterval>0，则等待pullInterval好秒后将PullRequest放入到PullMessageService的pullRequestQueue中，该消息队列的下次拉取即将被激活，达到持续消息拉取，实现准实时拉取消息的效果

3.4消息拉取长轮询机制分析

rocketmq并没有真正的实现推模式，而是消费者主动向消息服务器拉取消息，推模式是循环向消息服务端发送消息拉取请求

如果消息为到达消费队列，如果不启用长轮询机制，服务端会等待shortPollingTimeMills时间后挂起再去判断消息是否到达消息队列，如果消息味道大则提示消息拉取客户端PULL_NOT_FOUND消息不存在

如果开启长轮询模式，会每5s轮询检查一次消息是否到达，同时一有新消息到达立马通知挂起线程再次验证新消息是否是自己感兴趣的消息，如果是则从commitlog文件提取消息返回给客户端，否则直到挂起超时，超时时间PUSH模式默认15s

消息拉取时服务端从commitlog未找到消息时的逻辑处理如下
PullMessageProcessor#processRequest

从commitLog没有获取到数据，触发轮询机制，由两个线程共同完成
1、PullRequestHoldService：每隔5s重试一次
2、DefaultMessageStore#ReputMessageService，每处理一次重新拉取，Thread.sleep(1)

3.5PullRequestHoldService

PullRequestHoldService#run

 @Override
    public void run() {
        log.info("{} service started", this.getServiceName());
        while (!this.isStopped()) {
            try {
                // 如果开启长轮询，每5s一次，判断新消息是否到达
                if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                    this.waitForRunning(5 * 1000);
                } else {
                    // 未开启长轮询，则默认等待1s再次尝试
                    this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
                }

                long beginLockTimestamp = this.systemClock.now();
                this.checkHoldRequest();
                long costTime = this.systemClock.now() - beginLockTimestamp;
                if (costTime > 5 * 1000) {
                    log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
                }
            } catch (Throwable e) {
                log.warn(this.getServiceName() + " service has exception. ", e);
            }
        }

        log.info("{} service end", this.getServiceName());
    }

总结：如果开启了长轮询机制，PullRequestHoldService线程会每隔5s被唤醒去尝试检测是否有新消息的到来，直到超时，如果被挂起，需要等待5s，消息拉取实时性比较差，

为了避免这种情况，引入另外一种机制，当消息到达时唤醒挂起线程，触发一次检查

3.6DefaultMessageStore$ReputMessageService详解

ReputMessageService线程主要是根据commitlog将消息转发到consumeQueue，Index等文件。现关注doReput方法关于长轮询相关实现。

总结

待续

欢迎分享，转载请注明来源：内存溢出

原文地址: https://outofmemory.cn/langs/730359.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-04-27
下一篇 2022-04-27

发表评论

登录后才能评论

评论列表（0条）

保存