rocketmq核心源码分析第六篇一消费者启动

rocketmq核心源码分析第六篇一消费者启动

文章目录

前言

消费者模块图轻量级pull消费者来源DefaultMQPullConsumer使用方式DefaultLitePullConsumer使用方式DefaultMQPushConsumer使用方式三种模式对比 正文

DefaultMQPushConsumer消费者实例化DefaultMQPushConsumer.start

DefaultMQPushConsumerImpl.start核心源码分析 消费者内部组件关系图总结

扩展点一消费流程简述

前言 消费者模块图

pull和push模式消费者轻量级pull
轻量级pull消费者来源

issue提问: 由于当前 DefaultMQPullConsumer 过于原子化，用户在使用 pullConsumer 时需要过多关注底层，所以恕我直言，如果可以为 RocketMQ 添加一个 lite pull consumer 就好了
issue解决: 产生LitePullConsumer

DefaultMQPullConsumer使用方式

@Before
public void init() throws Exception {
    pullConsumer = new DefaultMQPullConsumer(consumerGroup);
    pullConsumer.setNamesrvAddr("127.0.0.1:9876");
    pullConsumer.start();
    ...... 删除其他代码
}

@Test
public void testPullMessage_Success() throws Exception {
      ...... 删除其他代码
    实际使用需要获取MessageQueue集合
    MessageQueue messageQueue = new MessageQueue(topic, brokerName, 0);
    实际使用需要存offsetstore获取进度
    PullResult pullResult = pullConsumer.pull(messageQueue, "*", 1024, 3);
     ...... 删除其他代码
}

DefaultLitePullConsumer使用方式

区分固定消费messageQueue还是订阅消费messageQueue

private DefaultLitePullConsumer createSubscribeLitePullConsumer() throws Exception {
    DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(consumerGroup + System.currentTimeMillis());
    litePullConsumer.setNamesrvAddr("127.0.0.1:9876");
    litePullConsumer.subscribe(topic, "*");
    litePullConsumer.start();
    return litePullConsumer;
}
assign指定消费Mq
litePullConsumer.assign(Collections.singletonList(messageQueue));
List result = litePullConsumer.poll();

订阅消费Mq
litePullConsumer.setPollTimeoutMillis(20 * 1000);
List result = litePullConsumer.poll();

DefaultMQPushConsumer使用方式

DefaultMQPushConsumer consumer = new  DefaultMQPushConsumer("please_rename_unique_group_name_4");
consumer.subscribe("Topic", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List msgs,
        ConsumeConcurrentlyContext context) {
        System.out.print(msgs);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

三种模式对比 名称特点DefaultMQPushConsumerpush模式 消费需要MessageListenerDefaultMQPullConsumerpull模式 消费需要业务自定义,通过自定义messagequeue拉取消息DefaultLitePullConsumerpull模式 消费需要业务自定义,通过再平衡拉取消息,获取assign指定messagequeue拉取消息 正文 DefaultMQPushConsumer消费者实例化

分配队列算法[所有节点采用无状态算法,即输入相同,输出相同]完成当前消费者订阅topic的messagequeue选择设置defaultMQPushConsumerImpl 负责消息的消费,拉取,负载均衡

  public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
                                 AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
        this.consumerGroup = consumerGroup;
        this.namespace = namespace;
        分配队列算法[所有节点采用无状态算法继续如何消费]
        this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
        defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
    }

设置rpcHook[钩子函数]

public DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook) {
        this.defaultMQPushConsumer = defaultMQPushConsumer;
        this.rpcHook = rpcHook;
        this.pullTimeDelayMillsWhenException = defaultMQPushConsumer.getPullTimeDelayMillsWhenException();
    }

DefaultMQPushConsumer.start

消息消费者启动消息轨迹启动

public void start() throws MQClientException {
    setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
    消息消费者启动
    this.defaultMQPushConsumerImpl.start();
    消息轨迹启动
    if (null != traceDispatcher) {
        try {
            traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
        } catch (MQClientException e) {
            log.warn("trace dispatcher start failed ", e);
        }
    }
}

DefaultMQPushConsumerImpl.start核心源码分析

构建订阅信息 集群模式下同时订阅重试主题[正常主题-> 定时主题-> 重试主题]构建mQClientFactory通信层构建rebalanceImpl负载均衡为rebalanceImpl设置无状态算法实现AllocateMessageQueueStrategy构建pullAPIWrapper消息拉取构建并加载offsetStore消息消费进度构建顺序或并发MessageListener启动消息消费线程mQClientFactory注册DefaultMQPushConsumerImpl初始rebalanceImpl.topicSubscribeInfoTable消费者topic订阅信息首次执行负载均衡

public synchronized void start() throws MQClientException {
    switch (this.serviceState) {
        case CREATE_JUST:
            ...... 删除其他代码
            检查配置信息，
            ---- 主要检查消费者组（consumeGroup）、消息消费方式（messageModel）、消息消费开始偏移量（consumeFromWhere）、
            ---- 消息队列分配算法（AllocateMessageQueueStrategy）、订阅消息主题（Map 定时主题     -> 重试主题]
            this.copySubscription();
            消息消费模式为集群模式，并且当前的实例名为 DEFAULT，替换为当前客户端进程的PID
            ...... 删除实例名处理
            获取mQClientAPIImpl底层通信实例
            this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
            载均衡相关实现
            this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
            this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
            设置负载策略 AllocateMessageQueueAveragely
            this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
            this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
            pullAPIWrapper，消息拉取API封装类 推拉的本质都是拉取
            this.pullAPIWrapper = new PullAPIWrapper(
                mQClientFactory,
                this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
            this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

            ...... 删除消息消费进度实现的选择[本地或远程]
            加载消息消费进度。
            this.offsetStore.load();

            判断是顺序消费还是并发消费
            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();
            向通信层注册消费者
            boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
            ...... 删除注册失败代码
            启动 再平衡线程 启动拉取线程
            mQClientFactory.start();
            log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
            this.serviceState = ServiceState.RUNNING;
            break;
        ...... 删除其他代码   
        default:
            break;
    }
    更新订阅信息 会触发rebalance和生产者的TopicPublishInfo更新
    this.updateTopicSubscribeInfoWhenSubscriptionChanged();
    检测broker状态
    this.mQClientFactory.checkClientInBroker();
    发送心跳包
    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    唤醒重新负载;重新负载每隔20s执行[======这里就是调用rebalanceImpl开始实现消费的地方]
    在由其rebalance线程调用 doRebalance
    this.mQClientFactory.rebalanceImmediately();
}

消费者内部组件关系图

defaultMQPushConsumer持有DefaultMQPushConsumerImplDefaultMQPushConsumerImpl持有rebalanceimplDefaultMQPushConsumerImpl持有pullAPIWrapperDefaultMQPushConsumerImpl持有messageListenerDefaultMQPushConsumerImpl持有offsetStoreDefaultMQPushConsumerImpl持有各组件任务的相关线程

总结

消费者涉及到订阅信息,负载均衡,拉取消息,消费消息等等[参见消息消费章节]负载均衡的分配策略由AllocateMessageQueueStrategy实现offsetStore区分本地和远程对应消费模式区分广播和集群消费区分并发和顺序负载均衡,拉取消息,消费消息均处于不同线程中数据相互协作,任务互相隔离 扩展点一消费流程简述

MQClientInstance通过监听consumer订阅topic变化触发rebalanceImpl进行负载均衡调整rebalanceImpl必要时构建PullRequest消息拉取服务pullmessageservice通过PullRequest向broker拉取消息拉取的消息放于内存中交由consumeMessageService消费拉取消息时同时上传当前offsetStore消费进度信息