RocketMQ - 消费者概述

rocketmq,消费者,概述 · 浏览次数 : 270

小编点评

**消费流程** **消费者组** * 一个逻辑概念,用于指定多个主题的消费者。 * 一个消费者组可以订阅多个主题。 * 订阅关系:一个主题的某一个标签,与消费者组名匹配。 **RocketMQ 消费模式** * 支持集群消费模式和广播消费模式。 * 集群消费模式:使用集群中的消费者实例消费主题上的消息。 * 每个消费者实例消费消息数等于生产消息数除以消费者实例数。 * 广播消费模式:广播所有主题上的消息给所有消费者实例。 **消费者实例** *多个消费者实例,每个实例可订阅多个主题。 * 每个消费者实例可设置不同的标签,以指定消费哪些主题。 **订阅关系不一致** * 订阅关系不一致会导致消费消息紊乱。 * 每个消费者实例只能订阅其所在的主题的标签。 **重试机制** * 在消费消息失败时,生产者可以重新手动发消息或将消息保存到重试队列中。 * 重试队列中的消息可进行多次重试。 **死信队列** * 当消息消费失败时,消息会被发送到死信队列中。 * 死信队列中的消息无法被正常消费,但可以用于处理特殊情况。 **其他** * RocketMQ 的消费流程包含 3 个阶段:正常消费、重试消费和死信。 * 每个消费者实例可以设置消费的策略,例如是否使用重试机制。 * 消费者组可以根据需要设置消费策略。

正文

消费流程

image
消费者组: 一个逻辑概念,在使用消费者时需要指定一个组名。一个消费者组可以订阅多个Topic。
消费者实例: 一个消费者组程序部署了多个进程,每个进程都可以称为一个消费者实例。
订阅关系: 一个消费者组订阅一个 Topic 的某一个 Tag,这种记录被称为订阅关系。RocketMQ规定消费订阅关系(消费者组名-Topic-Tag)必须一致——在此,笔者想提醒读者,一定要重视这个问题,一个消费者组中的实例订阅的Topic和Tag必须完全一致,否则就是订阅关系不一致。订阅关系不一致会导致消费消息紊乱。

消费模式

RocketMQ目前支持集群消费模式和广播消费模式,其中集群消费模式使用最为广泛

集群消费模式

在同一个消费者组中的消费者实例,是负载均衡(策略可以配置)地消费Topic中的消息,假如有一个生产者(Producer)发送了 120 条消息,其所属的 Topic 有 3 个消费者(Consumer)组,每个消费者组设置为集群消费,分别有2个消费者实例,如图所示。
image
Consumer Group A 的两个实例 Consumer Instance A1 和 Consumer Instance A2 分别负载均衡地消费60条消息。由此我们可以得出使用负载均衡策略时,每个消费者实例消费消息数=生产消息数/消费者实例数,在本例中是60=120/2

适用场景: 目前大部分场景都适合集群消费模式,RocketMQ 的消费模式默认是集群消费。比如异步通信、削峰等对消息没有顺序要求的场景都适合集群消费。因为集群模式的消费进度是保存在Broker端的,所以即使应用崩溃,消费进度也不会出错。

广播消费模式

广播消费,顾名思义全部的消息都是广播分发,即消费者组中的全部消费者实例将消费整个 Topic 的全部消息。比如,有一个生产者生产了 120 条消息,其所属的 Topic 有 3个消费者组,每个消费者组设置为广播消费,分别有两个消费者实例,如图所示。
image
Consumer Group A 的两个实例 Consumer Instance A1 和 Consumer Instance A2 分别消费120条消息。整个消费者组收到消息120×2=240条。由此我们可以得出广播消费时,每个消费者实例的消费消息数=生产者生产的消息数,整个消费者组中所有实例消费消息数=每个消费者实例消费消息数×消
费者实例数,本例中是240=120×2

适用场景: 广播消费比较适合各个消费者实例都需要通知的场景,比如刷新应用服务器中的缓存,如图所示。

image
生产者发一个刷新缓存的广播消息,消费者组如果设置为广播消费,那么每个应用服务中的消费者都可以消费这个消息,也都能刷新缓存。

广播消费的消费进度保存在客户端机器的文件中。如果文件弄丢了,那么消费进度就丢失了,可能会导致部分消息没有消费

可靠消费

RocketMQ是一种十分可靠的消息队列中间件,消费侧通过重试-死信机制、Rebalance机制等多种机制保证消费的可靠性。

重试-死信机制

我们假设有一个场景,在消费消息时由于网络不稳定导致一条消息消费失败。此时是让生产者重新手动发消息呢,还是自己做数据补偿?

横向看,RocketMQ的消费过程分为 3个阶段:正常消费、重试消费和死信。在引进了正常Topic、重试队列、死信队列后,消费过程的可靠性提高了。RocketMQ的消费流程如图所示
image

正常Topic: 正常消费者订阅的Topic名字。
重试Topic: 如果由于各种意外导致消息消费失败,那么该消息会自动被保存到重试Topic中,格式为“%RETRY%消费者组”,在订阅的时候会自动订阅这个重试Topic。

进入重试队列的消息有16次重试机会,每次都会按照一定的时间间隔进行。RocketMQ 认为消费不是一锤子买卖,可能由于各种偶然因素导致正常消费失败,只要正常消费或者重试消费中有一次消费成功,就算消费成功
image

死信Topic: 死信Topic名字格式为“%DLQ%消费者组名”。如果正常消费1次失败,重试16次失败,那么消息会被保存到死信Topic中,进入死信Topic的消息不能被再次消费。RocketMQ认为,如果17次机会都失败了,说明生产者发送消息的格式发生了变化,或者消费服务出现了问题,需要人工介入处理。

Rebalance机制

Rebalance(重平衡)机制,用于在发生Broker掉线、Topic扩容和缩容、消费者扩容和缩容等变化时,自动感知并调整自身消费,以尽量减少甚至避免消息没有被消费。后面会详细讲述Rebalance的过程。

死信队列

当一条消息初次消费失败,消息队列RocketMQ版会自动进行消息重试,达到最大重试次数后,若消费依然失败,则表明消费者在正常情况下无法正确地消费该消息。此时,消息队列RocketMQ版不会立刻将消息丢弃,而是将其发送到该消费者对应的特殊队列中。

在消息队列RocketMQ版中,这种正常情况下无法被消费的消息称为死信消息(Dead-Letter Message),存储死信消息的特殊队列称为死信队列(Dead-Letter Queue)

死信消息具有以下特性:

  • 不会再被消费者正常消费。
  • 有效期与正常消息相同,默认为3天,3天后会被自动删除。因此,请在死信消息产生后的3天内及时处理。

死信队列具有以下特性:

  • 一个死信队列对应一个Group ID, 而不是对应单个消费者实例。
  • 如果一个Group ID未产生死信消息,消息队列RocketMQ版不会为其创建相应的死信队列。
  • 一个死信队列包含了对应Group ID产生的所有死信消息,不论该消息属于哪个Topic。

与RocketMQ - 消费者概述相似的内容:

RocketMQ - 消费者概述

消费流程 消费者组: 一个逻辑概念,在使用消费者时需要指定一个组名。一个消费者组可以订阅多个Topic。 消费者实例: 一个消费者组程序部署了多个进程,每个进程都可以称为一个消费者实例。 订阅关系: 一个消费者组订阅一个 Topic 的某一个 Tag,这种记录被称为订阅关系。RocketMQ规定消费

【主流技术】聊一聊消息队列 RocketMQ 的基本结构与概念

RocketMQ 是阿里巴巴在 2012 年开源的分布式消息中间件,目前已经捐赠给 Apache 软件基金会,并于 2017 年 9 月 25 日成为 Apache 的顶级项目。 作为经历过多次阿里巴巴双十一这种“超级工程”的洗礼并有稳定出色表现的国产中间件,以其高性能、低延时和高可靠等特性近年来...

RocketMQ - 消费者启动机制

RocketMQ客户端中有两个独立的消费者实现类:org.apache.rocketmq.client.consumer.DefaultMQPullConsumer 和 org.apache.rocketmq.client.consumer.DefaultMQPushConsumer Default

RocketMQ - 消费者Rebalance机制

客户端是通过Rebalance服务做到高可靠的。当发生Broker掉线、消费者实例掉线、Topic 扩容等各种突发情况时,消费者组中的消费者实例是怎么重平衡,以支持全部队列的正常消费的呢? RebalancePullImpl 和 RebalancePushImpl 两个重平衡实现类,分别被 Defa

RocketMQ - 消费者进度保存机制

RocketMQ设计了远程位点管理和本地位点管理两种位点管理方式。集群消费时,位点由客户端提交给Broker保存,具体实现代码在RemoteBrokerOffsetStore.java文件中;广播消费时,位点保存在消费者本地磁盘上,实现代码在LocalFileOffsetStore.java文件中

RocketMQ - 消费者消费方式

RocketMQ的消费方式包含Pull和Push两种 Pull方式:用户主动Pull消息,自主管理位点,可以灵活地掌控消费进度和消费速度,适合流计算、消费特别耗时等特殊的消费场景。缺点也显而易见,需要从代码层面精准地控制消费,对开发人员有一定要求。 在 RocketMQ 中org.apache.ro

RocketMQ消费者是如何负载均衡的

摘要:RocketMQ 支持两种消息模式:集群消费( Clustering )和广播消费( Broadcasting )。 本文分享自华为云社区《一文讲透RocketMQ消费者是如何负载均衡的》,作者:勇哥java实战分享。 RocketMQ 支持两种消息模式:集群消费( Clustering )和

RocketMQ - 生产者最佳实践总结

相对消费者而言,生产者的使用更加简单,一般关注消息类型、消息发送方法和发送参数,即可正常使用RocketMQ发送消息 常用消息类型 | 消息类型 | 优点 | 缺 点 | 备注 | | | | | | | 普通消息(并发消息) | 性能最好。单机TPS的级别为100 000 | 消息的生产和消费都无

【RocketMQ】Rebalance负载均衡总结

消费者负载均衡,是指为消费组下的每个消费者分配订阅主题下的消费队列,分配了消费队列消费者就可以知道去消费哪个消费队列上面的消息,这里针对集群模式,因为广播模式,所有的消息队列可以被消费组下的每个消费者消费不涉及负载均衡,而集群模式一个消息队列同一时间只能分配给组内的一个消费者进行消费。 Rocket

【RocketMQ】消息的消费总结

消费者从Broker拉取到消息之后,会将消息提交到线程池中进行消费,RocketMQ消息消费是批量进行的,如果一批消息的个数小于预先设置的批量消费大小,直接构建消费请求ConsumeRequest将消费请求提交到线程池处理,否则需要分批构建进行提交。 消息消费 在消息被提交到线程池后进行处理时,会调