开发者社区> 小郭的技术笔记> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

【消息中间件】默认的RocketMQ消息消费者是如何启动的?(上)

简介: 在当下的分布式服务中,消息队列中间件是一个解决服务之间耦合的利器,今天我们来瞧一瞧开源的RocketMQ消息中间件,他的消费端是如何启动的,以及在使用他的过程中有哪些配置。
+关注继续查看

前言

大家好,我是小郭,在当下的分布式服务中,消息队列中间件是一个解决服务之间耦合的利器,今天我们来瞧一瞧开源的RocketMQ消息中间件,他的消费端是如何启动的,以及在使用他的过程中有哪些配置。

启动流程图

image

Push和Pull的区别

Apache RocketMQ在消费者服务中,为我们提供了Push模式也提供了Pull模式

那他们主要有什么区别呢?

  • Push是服务端主动推送消息给客户端,优点是及时性较好,但如果客户端没有做好流控,一旦服务端推送大量消息到客户端时,就会导致客户端消息堆积甚至崩溃。
  • Pull是客户端需要主动到服务端取数据,优点是客户端可以依据自己的消费能力进行消费,但拉取的频率也需要用户自己控制,拉取频繁容易造成服务端和客户端的压力,拉取间隔长又容易造成消费不及时。

BROADCASTING 和 CLUSTERING 模式的区别

BROADCASTING(广播模式):当使用广播消费模式时,RocketMQ 会将每条消息推送给消费组所有的消费者,保证消息至少被每个消费者消费一次。

CLUSTERING(集群模式):当使用集群消费模式时,RocketMQ 认为任意一条消息只需要被消费组内的任意一个消费者处理即可。

Consumer启动流程

我们直接跑一个官方提供的Demo,大家也可以去官网上去下载源码

public static void main(String[] args) throws InterruptedException, MQClientException {
    /*
     * Instantiate with specified consumer group name.
     * 消费者模式有两种 推和拉
     */
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
    consumer.setNamesrvAddr("127.0.0.1:9876");
    /*
     * Specify where to start in case the specific consumer group is a brand-new one.
     * 指定消费从哪里开始
     */
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    /*
     * Subscribe one more topic to consume.
     * 设置监听主题以及过滤条件
     */
    consumer.subscribe("TopicTest999", "*");
    /*
     *  Register callback to execute on arrival of messages fetched from brokers.
     *  注册消息监听器
     */
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            //System.out.println("待消费条数:"+ msgs.size());
            LOGGER.info("Receive New Messages : {}", Thread.currentThread().getName());
            /*try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/
            LOGGER.info("success");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    /*
     *  Launch the consumer instance.
     */
    consumer.start();
    System.out.printf("Consumer Started.%n");
}

前置设置

  1. 指定namesrvAddr地址
consumer.setNamesrvAddr("127.0.0.1:9876");
  1. 指定消费从哪里开始
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  • CONSUME_FROM_LAST_OFFSET 从最新的消息开始消费
  • CONSUME_FROM_FIRST_OFFSET 从最新的位点开始消费
  • CONSUME_FROM_TIMESTAMP 从指定的时间戳开始消费
  1. 指定负载均衡策略
  • AllocateMessageQueueAveragely:平均连续分配算法。
  • AllocateMessageQueueAveragelyByCircle:平均轮流分配算法。
  • AllocateMachineRoomNearby:机房内优先就近分配。
  • AllocateMessageQueueByConfig:手动指定,这个通常需要配合配置中心,在消费者启动时,首先先创建 AllocateMessageQueueByConfig 对象,然后根据配置中心的配置,再根据当前的队列信息,进行分配,即该方法不具备队列的自动负载,在 Broker 端进行队列扩容时,无法自动感知,需要手动变更配置。
// AllocateMessageQueueByConfig
AllocateMessageQueueByConfig allocateMessageQueueByConfig = new AllocateMessageQueueByConfig();
        MessageQueue messageQueue = new MessageQueue();
        messageQueue.setBrokerName("broker-a");
        messageQueue.setQueueId(2);
        messageQueue.setTopic("TopicTest");
allocateMessageQueueByConfig.setMessageQueueList(Collections.singletonList(messageQueue));
consumer.setAllocateMessageQueueStrategy(allocateMessageQueueByConfig);
  • AllocateMessageQueueByMachineRoom:消费指定机房中的队列,该分配算法首先需要调用该策略的 setConsumeridcs(Set consumerIdCs) 方法,用于设置需要消费的机房,将刷选出来的消息按平均连续分配算法进行队列负载。
// AllocateMessageQueueByMachineRoom    
AllocateMachineRoomNearby.MachineRoomResolver machineRoomResolver = new AllocateMachineRoomNearby.MachineRoomResolver() {
            // Broker部署
            @Override
            public String brokerDeployIn(MessageQueue messageQueue) {
                System.out.println(messageQueue.getBrokerName().split("-")[0]);
                return messageQueue.getBrokerName().split("-")[0];
            }
            // 消费端部署
            @Override
            public String consumerDeployIn(String clientID) {
                    System.out.println(clientID.split("-")[0]);
                return clientID.split("-")[0];
            }
        };
consumer.setAllocateMessageQueueStrategy(new AllocateMachineRoomNearby(new AllocateMessageQueueAveragely(), machineRoomResolver));
  1. 设置监听主题以及过滤条件

Tag过滤,用于对某个Topic下的消息进行分类,

消息发送到名称为TopicTest999的Topic中,被各个不同的系统所订阅,我们可以利用Tag来区分

consumer.subscribe("TopicTest999", "order");
consumer.subscribe("TopicTest999", "user");
  1. 注册消息监听器

注册消息监听器的目的就是为了接收消息,RocketMQ本身为我们提供了两种模式

  • 并发消费
consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                LOGGER.info("Receive New Messages : {}", Thread.currentThread().getName());
                /*try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }*/
                LOGGER.info("success");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
  • 顺序消费
consumer.registerMessageListener(new MessageListenerOrderly() {
            AtomicLong consumeTimes = new AtomicLong(0);
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    return ConsumeOrderlyStatus.SUCCESS;
                } else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

他们主要的区别继承MessageListener接口的实现

除了这一些重要的参数以外,RocketMQ为我们提供了其他非常丰富的配置,我总结在了下图

image

启动流程源码跟踪

需要注意的是,在配置后我们才能去调用启动方法

1. 设置消费者分组后,DefaultMQPushConsumer调用start()启动消费者

入口:org.apache.rocketmq.client.consumer.DefaultMQPushConsumer#start

@Override
public void start() throws MQClientException {
    // step 1 设置消费者分组
    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);
        }
    }
}

2. 根据serviceState状态启动消费者,当服务未创建时,才能启动成功

在这里主要做了五件事

  1. 检查核心参数是否都配置了
private void checkConfig() throws MQClientException {
         // 检查消费者组,是否满足条件
        Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup());
        if (null == this.defaultMQPushConsumer.getConsumerGroup()) {
            ...
        }
        if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) {
            ...
        }
        if (null == this.defaultMQPushConsumer.getMessageModel()) {
            ...
        }
        if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) {
            ...
        }
        // allocateMessageQueueStrategy
        // subscription
        // messageListener
        // consumeThreadMin
        // consumeThreadMax
        // consumeConcurrentlyMaxSpan
        // pullThresholdForQueue
        // pullThresholdForTopic
        // pullThresholdSizeForQueue
        // pullInterval
        // consumeMessageBatchMaxSize
        // pullBatchSize

主要是进行了参数配置的校验,如果一些参数设置不合理的,在这里就会抛出异常,终止了消费者服务的启动,这里的配置对后面的使用会产生一定的影响,所以我们在配置的时候需要更加的谨慎

  1. 复制订阅信息,生成重试主题
private void copySubscription() throws MQClientException {
    try {
        Map<String, String> sub = this.defaultMQPushConsumer.getSubscription();
        if (sub != null) {
            for (final Map.Entry<String, String> entry : sub.entrySet()) {
                final String topic = entry.getKey();
                final String subString = entry.getValue();
                SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(topic, subString);
                // 更新内部订阅关系
                this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
            }
        }
        if (null == this.messageListenerInner) {
            this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();
        }
        // 默认情况下我们是CLUSTERING模式
        switch (this.defaultMQPushConsumer.getMessageModel()) {
            case BROADCASTING:
                break;
            case CLUSTERING:
                // 创建重试主题
                final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
                SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(retryTopic, SubscriptionData.SUB_ALL);
                // 将重试主题放入订阅关系容器中
                this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
                break;
            default:
                break;
        }
    } catch (Exception e) {
        throw new MQClientException("subscription exception", e);
    }
}

获取配置的订阅关系,因为setSubscription()方法已经被作废,subscription都是为空的,在下面他会去维护一个subscriptionInner

protected final ConcurrentMap<String /* topic */, SubscriptionData> subscriptionInner =
        new ConcurrentHashMap<String, SubscriptionData>();

subscriptionInner保存了我们在前置配置的时候插入的订阅关系

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
《万亿级数据洪峰下的消息引擎——Apache RocketMQ》电子版地址
万亿级数据洪峰下的消息引擎——Apache RocketMQ
0 0
RocketMQ Schema——让消息成为流动的结构化数据
RocketMQ Schema 提供了对消息的数据结构托管服务,同时为原生客户端提供了较为丰富的序列化/反序列化 SDK ,补齐了 RocketMQ 在数据治理和业务上下游解耦方面的短板,让数据成为流动的结构化数据,那么快来了解下实现原理吧~
0 0
简述RocketMQ消息拉取过程【二】
简述RocketMQ消息拉取过程【二】
0 0
简述RocketMQ消息拉取过程【一】
简述RocketMQ消息拉取过程【一】
0 0
RocketMQ消息生产者是如何选择Broker的
RocketMQ消息生产者是如何选择Broker的
0 0
同步异步调用,并谈谈消息队列mq;RocketMQ发送消息和消费消息测试类
同步调用优点: 时效性强,打电话、直播,很快可以得到结果 同步调用的问题:
0 0
弥补延时消息的不足,RocketMQ 基于时间轮算法实现了定时消息!
弥补延时消息的不足,RocketMQ 基于时间轮算法实现了定时消息!
0 0
10 张图告诉你 RocketMQ 是怎样保存消息的
10 张图告诉你 RocketMQ 是怎样保存消息的
0 0
5 张图带你彻底理解 RocketMQ 轨迹消息
5 张图带你彻底理解 RocketMQ 轨迹消息
0 0
3 张图带你彻底理解 RocketMQ 事务消息
3 张图带你彻底理解 RocketMQ 事务消息
0 0
+关注
小郭的技术笔记
掘金平台-优秀创作者 专注Java后端、中间件、架构方向
文章
问答
文章排行榜
最热
最新
相关电子书
更多
万亿级数据洪峰下的消息引擎——Apache RocketMQ
立即下载
万亿级数据洪峰下的消息引擎-Apache RocketMQ
立即下载
行业实践:RocketMQ 业务集成典型行业应用和实践
立即下载