分布式消息队列RocketMQ源码分析之3 -- Consumer负载均衡机制 -- Rebalance
2017-04-21 21:44
1091 查看
同Kafka一样,RocketMQ也需要探讨一个问题:如何把一个topic的多个queue分摊给不同的consumer,也就是负载均衡问题。
在讨论这个问题之前,我们先看一下Client的整体架构。
从下图可以看出以下几点:
(1)Producer与Consumer的共同逻辑,封装在MQClientInstance,MQClientAPIImpl, MQAdminImpl这3个蓝色的类里面。所谓共同的逻辑,比如定期更新NameServer地址列表,定期更新TopicRoute,发送网络请求等。
(2)Consumer有2种,Pull和Push。下面会详细讲述这2者的区别。
同Kafka一样,RocketMQ也有Consumer Group的概念。参见Kafka http://blog.csdn.net/chunlongyu/article/details/52663090#t1
缺省的,RocketMQ和Kafka采用的是同样的策略:同1个Consumer Group的多个Consumer,是分摊,也就是负载均衡;多个Consumer Group之间,是Pub/Sub模式。
但RocketMQ对此还做了扩展,允许同1个Consumer Group内部,也可以是广播模式。具体到代码里面,就是:
2
3
4
5
6
7
8
9
10
11
12
13
1
2
3
4
5
6
7
8
9
10
11
12
13
缺省的,Consumer的MessageModel就是CLUSTERING模式,也就是同1个Consumer Group内部,多个Consumer分摊同1个topic的多个queue,也就是负载均衡。
如果你把MessageModel改成BROADCAST模式,那同1个Consumer Group内部也变成了广播,此时ConsumerGroup其实就没有区分的意义了。此时,不管是1个Consumer Group,还是多个Consumer Group,对同1个topic的消息,都变成了广播。
下面我们先看一下pull和push的最基本用法:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
2
3
4
5
6
7
8
9
10
11
12
13
14
1
2
3
4
5
6
7
8
9
10
11
12
13
14
从上面的代码我们可以看出,pull和push用法上的基本差别就是:pull是客户端主动去拉取消息,push是注册了一个回调,当有新消息,该回调被调用。
但这还不是2者的最大区别,最大区别是:在pull里面,所有MessageQueue是向我们暴露的,我们需要自己去手动遍历所有的queue;而在push里面,我们只指定了订阅的topic,而MessageQueue是向我们隐藏的,在其内部做了“负载均衡”。
而上面的pull的代码,我们手动遍历了所有的queue,没有负载均衡!!!
那对于Pull模式,如何做负载均衡呢??
在MQPullConsumer这个类里面,有一个MessageQueueListener,它的目的就是当queue发生变化的时候,通知Consumer。也正是这个借口,帮助我们在Pull模式里面,实现负载均衡。
注意,这个接口在MQPushConsumer里面是没有的,那里面有的是上面代码里的MessageListener。
2
3
4
5
6
1
2
3
4
5
6
有了这个Listener,我们就可以动态的知道当前的Consumer分摊到了几个MessageQueue。然后对这些MessageQueue,我们可以开个线程池来消费。
幸运的是,RocketMQ给我们提供了一个工具类,MQPullConsumerScheduleService,帮助我们在pull模式下,实现负载均衡。
类似Push模式,在这个代码里面,我们也只指定了topic,而不像上面简陋的pull版本,要自己遍历所有的messageQueue。其内部帮我们做了负载均衡。
其使用代码如下:
在讨论这个问题之前,我们先看一下Client的整体架构。
Producer与Consumer类体系
从下图可以看出以下几点: (1)Producer与Consumer的共同逻辑,封装在MQClientInstance,MQClientAPIImpl, MQAdminImpl这3个蓝色的类里面。所谓共同的逻辑,比如定期更新NameServer地址列表,定期更新TopicRoute,发送网络请求等。
(2)Consumer有2种,Pull和Push。下面会详细讲述这2者的区别。
Consumer Group的Clustering模式与Pub/Sub模式
同Kafka一样,RocketMQ也有Consumer Group的概念。参见Kafka http://blog.csdn.net/chunlongyu/article/details/52663090#t1缺省的,RocketMQ和Kafka采用的是同样的策略:同1个Consumer Group的多个Consumer,是分摊,也就是负载均衡;多个Consumer Group之间,是Pub/Sub模式。
但RocketMQ对此还做了扩展,允许同1个Consumer Group内部,也可以是广播模式。具体到代码里面,就是:
*/ public enum MessageModel { /** * broadcast */ BROADCASTING("BROADCASTING"), /** * clustering */ CLUSTERING("CLUSTERING"); //缺省取值 ... }1
2
3
4
5
6
7
8
9
10
11
12
13
1
2
3
4
5
6
7
8
9
10
11
12
13
缺省的,Consumer的MessageModel就是CLUSTERING模式,也就是同1个Consumer Group内部,多个Consumer分摊同1个topic的多个queue,也就是负载均衡。
如果你把MessageModel改成BROADCAST模式,那同1个Consumer Group内部也变成了广播,此时ConsumerGroup其实就没有区分的意义了。此时,不管是1个Consumer Group,还是多个Consumer Group,对同1个topic的消息,都变成了广播。
Pull Consumer 与 Push Consumer
Push的负载均衡
下面我们先看一下pull和push的最基本用法:public static void main(String[] args) throws MQClientException { DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5"); //指定Consumer Group consumer.start(); Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1"); //获取一个topic的所有MessageQueue for (MessageQueue mq : mqs) { System.out.printf("Consume from the queue: " + mq + "%n"); SINGLE_MQ: while (true) { try { PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32); //遍历所有queue,挨个调用pull System.out.printf("%s%n", pullResult); putMessageQueueOffset(mq, pullResult.getNextBeginOffset()); switch (pullResult.getPullStatus()) { case FOUND: break; case NO_MATCHED_MSG: break; case NO_NEW_MSG: break SINGLE_MQ; case OFFSET_ILLEGAL: break; default: break; } } catch (Exception e) { e.printStackTrace(); } } } consumer.shutdown(); }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public static void main(String[] args) throws InterruptedException, MQClientException { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1"); //指定Consumer Group consumer.subscribe("Jodie_topic_1023", "*"); //指定要消费的topic consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); consumer.registerMessageListener(new MessageListenerConcurrently() { //该topic的任何一个queue有新消息,该回调回被调用 @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n"); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); System.out.printf("Consumer Started.%n"); }1
2
3
4
5
6
7
8
9
10
11
12
13
14
1
2
3
4
5
6
7
8
9
10
11
12
13
14
从上面的代码我们可以看出,pull和push用法上的基本差别就是:pull是客户端主动去拉取消息,push是注册了一个回调,当有新消息,该回调被调用。
但这还不是2者的最大区别,最大区别是:在pull里面,所有MessageQueue是向我们暴露的,我们需要自己去手动遍历所有的queue;而在push里面,我们只指定了订阅的topic,而MessageQueue是向我们隐藏的,在其内部做了“负载均衡”。
而上面的pull的代码,我们手动遍历了所有的queue,没有负载均衡!!!
那对于Pull模式,如何做负载均衡呢??
Pull的负载均衡
在MQPullConsumer这个类里面,有一个MessageQueueListener,它的目的就是当queue发生变化的时候,通知Consumer。也正是这个借口,帮助我们在Pull模式里面,实现负载均衡。注意,这个接口在MQPushConsumer里面是没有的,那里面有的是上面代码里的MessageListener。
void registerMessageQueueListener(final String topic, final MessageQueueListener listener); public interface MessageQueueListener { void messageQueueChanged(final String topic, final Set<MessageQueue> mqAll, final Set<MessageQueue> mqDivided); }1
2
3
4
5
6
1
2
3
4
5
6
有了这个Listener,我们就可以动态的知道当前的Consumer分摊到了几个MessageQueue。然后对这些MessageQueue,我们可以开个线程池来消费。
MQPullConsumerScheduleService
幸运的是,RocketMQ给我们提供了一个工具类,MQPullConsumerScheduleService,帮助我们在pull模式下,实现负载均衡。类似Push模式,在这个代码里面,我们也只指定了topic,而不像上面简陋的pull版本,要自己遍历所有的messageQueue。其内部帮我们做了负载均衡。
其使用代码如下:
public static void main(String[] args) throws MQClientException { final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService("GroupName1"); scheduleService.setMessageModel(MessageModel.CLUSTERING); scheduleService.registerPullTaskCallback("TopicTest1", new PullTaskCallback() { @Override public void doPullTask(MessageQueue mq, PullTaskContext context) { MQPullConsumer consumer = context.getPullConsumer(); try { long offset = consumer.fetchConsumeOffset(mq, false); if (offset < 0) offset = 0; PullResult pullResult = consumer.pull(mq, "*", offset, 32); System.out.printf("%s%n", offset + "\t" + mq + "\t" + pullResult); switch (pullResult.getPullStatus()) { case FOUND: break; case NO_MATCHED_MSG: break; case NO_NEW_MSG: case OFFSET_ILLEGAL: break; default: break; } consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset()); context.setPullNextDelayTimeMillis(100); } catch (Exception e) { e.printStackTrace(); } } }); scheduleService.start(); }
相关文章推荐
- 分布式消息队列RocketMQ源码分析之3 -- Consumer负载均衡机制 -- Rebalance
- 分布式消息队列RocketMQ源码分析之4 -- Consumer负载均衡与Kafka的Consumer负载均衡之不同点
- 分布式消息队列RocketMQ源码分析之2 -- Broker与NameServer心跳机制
- 分布式消息队列RocketMQ源码分析之1 -- Topic路由数据结构解析 -- topicRoute与topicPublishInfo与queueId
- 分布式消息队列RocketMQ源码分析之1 -- Topic路由数据结构解析 -- topicRoute与topicPublishInfo与queueId
- 分布式消息队列RocketMQ源码分析之1 -- Topic路由数据结构解析 -- topicRoute与topicPublishInfo与queueId
- 分布式消息队列RocketMQ源码分析之3 -- Consumer负载均衡机制 -- Rebalance
- 源码分析RocketMQ之CommitLog消息存储机制
- 源码分析RocketMQ之消息ACK机制(消费进度)
- 源码分析RocketMQ之消费队列、Index索引文件存储结构与存储机制-上篇
- 源码分析RocketMQ之消息消费重试机制
- 分布式消息队列 RocketMQ 源码分析——高可用
- 【RabbitMQ学习记录】- 消息队列存储机制源码分析
- 分布式消息队列RocketMQ&Kafka -- 消息的“顺序消费”
- 分布式消息队列RocketMQ&Kafka -- 消息的“顺序消费”-- 一个看似简单的复杂问题
- (原创)Rocketmq分布式消息队列的部署与监控
- 分布式消息队列RocketMQ&Kafka -- 消息的“顺序消费”-- 一个看似简单的复杂问题
- 源码分析RocketMQ消息消费机制----消费端消息负载均衡机制与重新分布
- 源码分析RocketMQ文件清除机制
- 分布式消息队列RocketMQ&Kafka -- 消息的“顺序消费”-- 一个看似简单的复杂问题