RocketMQ 消费者监听模型 解析——图解、源码级解析

发布时间:2023-12-17 19:30

Java学习:Java从入门到精通总结

深入浅出RocketMQ设计思想:深入浅出RocketMQ设计思想

绝对不一样的职场干货:大厂最佳实践经验指南


最近更新:2022年9月1日

个人简介:通信工程本硕、Java程序员。做过科研paper,发过专利,优秀的程序员不应该只是CRUD

点赞 收藏 ⭐留言 都是我最大的动力!


文章目录

  • 消息接收模型
  • 消费端监听模型
    • 监听器设计模式
    • 注册监听器
    • 启动消费者
    • `ConsumeMessageConcurrentlyService`类

前面已经把RocketMQ发送消息的主要流程梳理了一遍,接下来我将介绍消费者消费消息的源码及流程,主要包含但不仅限于一下内容:

  • 消息消费模式
  • 消息消费的流程
  • 消费者负载均衡算法
  • 消费失败如何进行重试
  • 消息如何进行重新投递

消息接收模型

Consumer消费者组消费消息时,受限从注册中心里获取Broker服务器中Topic的队列的地址,然后根据负载均衡算法进行消费。

RocketMQ的消息推送方式有拉模式和推模式,但实际上推模式的底层也是用的拉模式。

消费者组会将所有MQ分均分配给所有的组内消费者,同一个组内的消费者不能同时监听多个Topic,否则可能由于负载均衡算法是的部分消息无法被消费。

如果消息消费失败则会进行重试,默认是重试6次,如果还是没有消费成功的话消息就会被放入死信队列。

RocketMQ 消费者监听模型 解析——图解、源码级解析_第1张图片


消费端监听模型

监听器设计模式

RocketMQ采用了设计模式中的监听器设计模式,类与类之间的继承关系如图:
RocketMQ 消费者监听模型 解析——图解、源码级解析_第2张图片
ConsumeMessageConcurrentlyService就是监听器的执行类


注册监听器

编写自定义消费端并绑定监听器的代码如下:

public static void main(String[] args) throws Exception{
    // 创建消费者对象,设置组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Group A");
    // 订阅消息主题target_topic下的消息,*表示所有消息
    consumer.subscribe("target_topic", "*");
    // 设置从上一次的位点开始消费
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    // 注册监听器
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        // 消费消息的方法
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    // 开启消费者
    consumer.start();
    System.out.println("Consumer Started");
}

步骤如下:

  1. 创建消费者,指定所在的组
  2. 订阅主题target_topic,并消费所有消息
  3. 设置从上一次消费到的位置继续进行消费
  4. 注册监听器MessageListenerConcurrently,定义消费消息的逻辑consumeMessage
  5. 启动消费者

启动消费者

Consumer注册完监听器后,此时还没有开始消费,当Consumer调用了start方法之后才开始进行消费前的各种准备工作。

    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();

                // 复制订阅数据
                this.copySubscription();

                // 将InstanceName设置为pid(数字)
                if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                    this.defaultMQPushConsumer.changeInstanceNameToPID();
                }

                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(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);

                // 封装拉取消息的API
                this.pullAPIWrapper = new PullAPIWrapper(
                        mQClientFactory,
                        this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

                // 存储消费进度。集群模式下消费进度保存在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();

                // 根据顺序监听还是并发监听创建相应的Service
                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();

                // 检查是否注册成功,如果之前有重复的group名称则失败
                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                if (!registerOK) {
                    this.serviceState = ServiceState.CREATE_JUST;
                    this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
                    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;
        }
        // 从注册中心获取TopicRouteData,更新TopicPublishInfo和MQ(周期性调用)
        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        this.mQClientFactory.checkClientInBroker();
        // 向TopicRouteData中所有的Broker发送心跳,注册Consumer/Producer信息到Broker上(周期性调用)
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        // 唤醒MQ均衡服务,之后开始拉区消息
        this.mQClientFactory.rebalanceImmediately();
    }

上面代码的执行流程如下:

1. 先判断服务状态,Consumer是否处于运行状态

  • CREATE_JUST:Consumer刚创建,还没开始运行
  • RUNNING:Consumer正在运行
  • SHUTDOWN_ALREADY:Consumer已经关闭
  • START_FAILED:Consumer启动失败

2. 如果Consumer刚刚创建,则进行一系列检查工作

  • 如果是刚刚创建Consumer,则先把状态位设置为START_FAILED,等初始化流程执行完毕后再设置为RUNNING
  • 检查接收消息相关配置,如:组名格式是否正确等等
  • 获取Consumer的唯一标识,格式为ip@instanceName
  • 设置负载均衡消费模式,指定负载均衡的组、消息消费的策略
  • 拉取消息API进行封装
  • 创建消费进度处理器,集群模式下消费进度应该保存在Broker上,组内的消费者要共享进度、广播模式下的消费进度保存在Consumer
  • 根据监听时顺序模式还是并发模式来创建相应的ConsumerService。消费就是这个类的实例来实现的



ConsumeMessageConcurrentlyService

public ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
                                    MessageListenerOrderly messageListener) {
    // 设置消息消费者对象
    this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
    // 设置注册监听器对象
    this.messageListener = messageListener;
    // 获取消息消费者
    this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
    // 获取消息消费者所在的组
    this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
    // 创建消费者队列
    this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();
    // 消费者线程池
    this.consumeExecutor = new ThreadPoolExecutor(
            this.defaultMQPushConsumer.getConsumeThreadMin(),
            this.defaultMQPushConsumer.getConsumeThreadMax(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.consumeRequestQueue,
            new ThreadFactoryImpl("ConsumeMessageThread_"));

    this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
}

消费者消费消息最终就是指定ConsumeMessageConcurrentlyService来进行消费:

  • MessageListenerOrderly顺序消息监听器,使用ConsumeMessageOrderlyService对象来进行消息消费
  • MessageListenerConcurrently并发消息监听器使用ConsumeMessageConcurrentlyService对象进行消息消费

ItVuer - 免责声明 - 关于我们 - 联系我们

本网站信息来源于互联网,如有侵权请联系:561261067@qq.com

桂ICP备16001015号