发布时间:2023-12-17 19:30
Java学习:Java从入门到精通总结
深入浅出RocketMQ设计思想:深入浅出RocketMQ设计思想
绝对不一样的职场干货:大厂最佳实践经验指南
最近更新:2022年9月1日
个人简介:通信工程本硕、Java程序员。做过科研paper,发过专利,优秀的程序员不应该只是CRUD
点赞 收藏 ⭐留言 都是我最大的动力!
前面已经把RocketMQ发送消息的主要流程梳理了一遍,接下来我将介绍消费者消费消息的源码及流程,主要包含但不仅限于一下内容:
Consumer消费者组消费消息时,受限从注册中心里获取Broker服务器中Topic
的队列的地址,然后根据负载均衡算法进行消费。
RocketMQ的消息推送方式有拉模式和推模式,但实际上推模式的底层也是用的拉模式。
消费者组会将所有MQ分均分配给所有的组内消费者,同一个组内的消费者不能同时监听多个Topic
,否则可能由于负载均衡算法是的部分消息无法被消费。
如果消息消费失败则会进行重试,默认是重试6次,如果还是没有消费成功的话消息就会被放入死信队列。
RocketMQ采用了设计模式中的监听器设计模式,类与类之间的继承关系如图:
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");
}
步骤如下:
target_topic
,并消费所有消息MessageListenerConcurrently
,定义消费消息的逻辑consumeMessage
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刚刚创建,则进行一系列检查工作
START_FAILED
,等初始化流程执行完毕后再设置为RUNNING
ip@instanceName
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
对象进行消息消费