rocketmq-笔记

rocketmq-笔记

起男 262 2023-09-11

rocketmq-笔记

基本概念

消息(message)

消息系统所传输信息的物理载体,生产和消费数据的最小单位,每条消息必须属于一个主题

主题(topic)

topic表示一类消息的集合,每个主题包含若干条消息,每条消息只能属于一个主题,是rocketmq进行消息订阅的单位

一个生产者可以同时发送多种topic的消息;而一个消费者只能对某种特定的topic感兴趣,即只可以订阅和消费一种topic消息

标签(tag)

为消息设置的标签,用于同一主题下区分不同类型的消息。来自同一业务单元的消息,可以根据不同业务目的在同一主题下设置不同标签。标签能够有效的保持代码的清晰度和连贯性,并优化rocketMQ提供的查询系统。消费者可以根据tag实现对不同子主题的不同消费逻辑,实现更好的扩展性

队列(queue)

存储消息的物理实体。一个topic中可以包含多个queue,每个queue中存放的就是该topic的消息。一个topic的queue也被称为一个topic中的消息分区

类似于kafka中的分区partition

消息标识(messageId/key)

rocketmq中每个消息拥有唯一的messageId,且可以携带具有业务标识的key,以方便对消息的查询。不过需要注意的是,messageId有两个:在生产者send()消息时会自动生成一个messageId(msgId),当消息到达broker后,broker也会自动生成一个messageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识

  • msgId:由producer端生成,其生成规则为:

    producerIp+进程pid+MessageClientIDSetter类的classLoader的hashCode+当前时间+AutomicInteger自增计数器

  • offsetMsgId:由broker端生成,其生成规则为:

    brokerIp+物理分区的offset(queue中的偏移量)

  • key:由用户指定的业务相关的唯一标识

系统架构

producer

消息生产者,负责生产消息。producer通过mq的负载均衡模块选择相应的broker集群队列进行消息投递,投递的过程支持快速失败并且低延迟

rocketmq中的消息生产者都是以生产者组(producer group)的形式出现的。生产者组是同一类生产者的集合,一个生产者组可以同时发送多个topic的消息

consumer

消息消费者,负责消费消息。一个消费者会从broker服务器中获取到消息,并对消息进行相关业务处理

rockermq中的消费者都是以消费者组(consumer group)的形式出现的。消费者组是同一类消费者的集合,这类consumer消费的是同一个topic的消息。消费者组使得在消息消费方面,实现负载均衡和容错的目标变得非常容易

消费者组中的consumer的数量应该小于等于订阅的topic中的queue的数量。如果超出queue数量,则多出的consumer不能消费消息

一个topic的消息可以被多个消费者组同时消费

nameServer

NameServer是一个broker与topic路由的注册中心,支持broker的动态注册与发现

主要功能为:

  • broker管理:接受broker集群的注册信息并且保存下来作为路由信息的基本数据;提供心跳检测机制,检测broker是否存活
  • 路由信息管理:每个NameServer中都保存着broker集群的整个路由信息和用于客户端查询的队列信息。producer和consumer通过NameServer可以获取整个比如broker集群的路由信息,从而进行消息的投递和消费

rocketmq 1.0和2.0版本中,依赖的仍是zookeeper。从3.0开始去掉了zookeeper的依赖,使用了自己的NameServer

broker

broker充当着消息中转角色,负责存储消息、转发消息。broker在rocketmq系统中负责接收并存储生产者发送过来的消息,同时为消费者的拉取请求作准备。broker同时也存储着消息相关的元数据,包括消费者组消费进度偏移量offset、主题、队列等

NameServer

路由注册

NameServer通常也是以集群的方式部署,不过,NameServer是无状态的,即NameServer集群中的各个节点间是无差异的,各节点间相互不进行信息通信。那各节点中的数据是如何进行数据同步的呢?在broker节点启动时,轮询NameServer列表,与每个NameServer节点建立长连接,发起注册请求。在每个NameServer内部维护者一个broker列表,用来动态存储broker的信息

这种NameServer的无状态方式

  • 优点:集群搭建简单
  • 缺点:对于broker,必须明确指出所有NameServer地址,否则未指出的将不会注册,也因为如此,NameServer并不能随便扩容。如果不重新配置broker,则新增的NameServer对broker来说是不可见的

broker节点为了证明自己是活着的,为了维护与NameServer间的长连接,会将最新的信息以心跳包的方式上报给NameServer,每30秒发送一次心跳。心跳包中包含BrokerId、Broker地址、Broker名称、Broker所属集群名称等。NameServer在接收到心跳包后,会更新心跳时间戳,记录这个broker的最新存活时间

路由剔除

由于broker关机、宕机或网络抖动等原因,NameServer没有收到Broker的心跳,NameServer可能会将其从Broker列表中剔除

NameServer中有一个定时任务,每隔10秒就会扫描一次Broker列表,查看每一个broker的最新心跳时间戳距离当前时间是否超过120秒,如果超过,则会判定broker失败,然后将其从broker列表中剔除

路由发现

rocketmq的路由发现采用的是pull模型。当topic路由信息出现变化时,NameServer不会主动推送给客户端,而是客户端定时拉取topic最新的路由。默认客户端每30秒会拉取一次最新的路由

push模型:推送模型。实时性好,但需要维护一个长连接,消耗资源较大

pull模型:拉取模型。存在的问题是,实时性差

long polling模型:长轮询模型。其是对push和pull模型的整合,充分利用了这两种模型的优势

客户端NameServer选择策略

客户端(producer和consumer)在配置时必须要写上NameServer集群的地址,那么客户端到底连接的是哪个NameServer节点呢?客户端首先会生成一个随机数,然后再与NameServer节点数量取模,此时得到的就是所要连接的节点索引,然后就会进行连接。如果连接失败,则会采用轮询策略,逐个尝试着去连接其它节点

工作流程

  1. 启动NameServer,NameServer启动后开始监听端口,等待broker、producer、consumer连接
  2. 启动broker时,broker会与所有NameServer建立并保持长连接,然后每30秒向NameServer定时发送心跳
  3. 发送消息前,可以先创建topic,创建topic时需要指定该topic要存储在哪些broker上,当然,在创建topic时也会将topic与broker的关系写入到NameServer中。不过,这步是可选的,也可以在发送消息时自动创建topic
  4. producer发送消息,启动时先根NameServer集群中的其中一台建立长连接,并从NameServer中获取路由信息,即当前发送的topic的queue与broker的地址(ip+port)的映射关系。然后根据算法策略从中选择一个queue,与队列所在的broker建立长连接从而向broker发送消息。当然,在获取到路由信息后,producer会首先将路由信息缓存到本地,再每30秒从NameServer更新一次路由信息
  5. consumer跟producer类似,跟其中一台NameServer建立长连接,获取其所订阅topic的路由信息,然后根据算法策略从路由信息中获取到其所要消费的queue,然后直接跟broker建立长连接,开始消费其中的消息。consumer在获取到路由信息后,同样也会每30秒从NameServer更新一次路由信息。不过不同于producer的是,consumer还会向broker发送心跳,以确保broker的存活状态

手动topic的创建模式:

  • 集群模式:该模式下创建的topic在集群中,所有broker中的queue数量是相同的
  • broker模式:该模式下创建的topic在集群中,每个broker的queue数量可以不同

自动创建topic时,默认采用的是broker模式,会为每个broker默认创建4个queue

读写队列

从物理上来讲读写队列是同一个队列。所以,不存在读写队列数据同步的问题。读写队列是逻辑上进行区分的概念。一般情况下,读写队列数量是相同的

如果创建topic时的写队列数量为8,读队列为4,此时系统会创建8个queue,分别为0-7。producer会将消息写入到这8个队列,但consumer只会消费0-3这4个queue中的消息,4-7中的消息是不会被消费到的

如果创建topic时的写队列数量为4,读队列为8,此时系统会创建8个queue,分别为0-7。producer会将消息写入到0-3这4个队列,但consumer只会消费0-7这8个queue中的消息,但是4-7中是没有消息的。此时假设consumer group中包含两个consumer,consumer1消费0-3,consumer2消费4-7。consumer2是没有消息可消费的

当读写队列数量不同时,总是有问题的,那么为什么这样设计?这样设计是为了方便topic的缩容。例如,原来topic中有16个queue,如何能使其queue缩容为8,还不会丢失消息?可以动态修改写队列数量为8,读队列数量不变。此时新的消息只能写入前8个队列,而消费的却是16个队列中的数据。当发现后8个队列中的消息消费完毕后,就可以再将读队列数量动态设置为8.整个缩容过程没有丢失任何消息

perm

用于设置当前创建topic的操作权限:

  • 2:只写
  • 4:只读
  • 6:读写

复制策略

复制策略是broker的master与slave间的数据同步方式。分为同步复制和异步复制:

  • 同步复制:消息写入master后,master会等待slave同步数据成功后才向producer返回成功ack
  • 异步复制:消息写入master后,master立即向producer返回成功ack,无需等待slave同步成功

刷盘策略

刷盘策略指的是broker中消息的落盘方式,即消息发送到broker内存后消息持久化到磁盘的方式。分为同步刷盘和异步刷盘:

  • 同步刷盘:当消息持久化到broker的磁盘后才算是消息写入成功
  • 异步刷盘:当消息写入到broker的内存后即表示消息写入成功,无需等待消息持久化到磁盘

消息写入到broker的内存,一般是写入到了pageCache

对于异步刷盘策略,消息会写入到pageCache后立即返回成功ack。但并不会立即做落盘操作,而是当pageCache到达一定量时会自动进行落盘

broker集群模式

单master

只有一个broker。这种方式也只能是在测试时使用,生成环境下不能使用,因为存在单点问题

多master

broker集群仅由多个master构成,不存在slave。同一topic的各个queue会平均分布在各个master节点上

  • 优点:配置简单,单个master宕机或重启维护对应用无影响,在磁盘配置为raid10时,即使机器宕机不可恢复情况下,由于raid10磁盘非常可靠,消息也不会丢失(异步刷盘丢失少量消息,同步刷盘一条不丢),性能最高

    如果没有配置raid磁盘阵列,一旦出现master宕机,则会发送大量消息丢失的情况

  • 缺点:单台机器宕机期间,这台机器上未被消费的消息在机器恢复前不可订阅(不可消费)消息实时性会受到影响

多master多slave(异步复制)

broker集群由多个master构成,每个master又配置了多个slave(在配置了raid磁盘阵列的情况下,一个master一般配置一个slave即可)。master与slave的关系是主备关系,即master负责处理消息的读写请求,而slave仅负责消息的备份和master宕机后的角色切换

异步复制策略,即消息写入master成功后,立即返回ack,无需等待slave同步成功。该模式下当master宕机后,可能会存在少量消息丢失的问题

多master多slave(同步双写)

该模式是多个master多slave的同步复制实现。所谓同步双写,指的是消息写入master成功后,master会等待slave同步数据成功后才返回ack,即master与slave都要写入成功才会返回ack,也即双写

该模式与异步复制模式比,优点是消息更安全,不存在丢失消息的情况。但单个消息的rt略高,从而导致性能略低(大约10%)

消息的生产

producer可以将消息写入到某broker中的某queue中,其过程如下:

  1. producer发送消息之前,会先向NameServer发出获取topic路由信息的请求
  2. NameServer返回该topic的路由表和broker列表
  3. producer根据代码中指定的queue选择策略,从queue列表中选出一个queue,用于后续存储消息
  4. producer对消息做一些特殊处理,例如,消息本身超过4m,则会进行压缩
  5. producer向选择的queue所在的broker发出rpc请求,将消息发送到queue

路由表:实际上是一个map,key为topic名称,value为一个QueueData实例列表

QueueData:一个topic在一个broker中的所有queue

broker列表:也是一个map。key为brokerName,value为BrokerData

BrokerData:一组brokerName相同的小集群(master-slave)。包含brokerName和一个map。map的key为brokerId,value为该broker的地址(brokerId为0为master,非0为slave)

queue选择算法

对于无序消息,其queue选择算法,也称为消息投递算法,常见的有两种:

  • 轮询算法:默认算法,该算法保证了每个queue可以均匀的获取到消息
  • 最小投递延迟算法:该算法会统计每次消息投递的时间延迟,然后根据统计出的结果将消息投递到时间延迟最小的queue。如果延迟相同,则采用轮询算法投递

消息的存储

rocketmq中的消息存储在本地文件系统中,这些相关文件默认在当前用户主目录下store目录中

  • abort:该文件在broker启动后会自动创建,正常关闭broker,该文件会自动消失。若在没有启动broker的情况下,发现这个文件是存在的,则说明之前broker的关闭是非正常的
  • checkpoint:其中存储着commitlog、consumequeue、index文件的最后刷盘时间戳
  • commitlog:其中存放着commitlog文件,而消息是写在commitlog文件中的
  • consumequeue:其中存放着consumequeue文件,队列就存放在这个目录中
  • index:其中存放着消息索引文件indexFile
  • config:存放着broker运行期间的一些配置数据
  • lock:运行期间使用到的全局资源锁

commitlog

commitlog目录中存放着很多的mappedFile文件,当前broker中的所有消息都是落盘到这些mappedFile文件中的。mappedFile文件最大为1g,文件名由20位十进制数构成,表示当前文件的第一条消息的起始位偏移量

注意:一个broker中仅包含一个commitlog目录,所有的mappedFile文件都是存放在该目录中的。即无论当前broker中存放着多少topic的消息,这些消息都是被顺序写入到了mappedFile文件中的。也就是说,这些消息在broker中存放时没有按照topic进行分类存放

消息单元

mappedFile文件内容由一个个的消息单元构成。每个消息单元中包含消息总长度MsgLen、消息的物理位置physicalOffset、消息体内容Body、消息体长度BodyLength、消息主题Topic、Topic长度TopicLength、消息生产者BornHost、消息发送时间戳BornTimestamp、消息所在的队列QueueId、消息在queue中存储的偏移量QueueOffset等近20余项相关属性

consumequeue

为了提高效率,会为每个topic在/store/consumequeue中创建一个目录,目录名称为topic名称,在该目录下,会再为每个queue创建一个目录,目录名为queueId。每个目录中存放着若干consumequeue文件,consumequeue文件是commitlog的索引文件,可以根据consumequeue定位到具体的消息

consumequeue文件名也是由20位数字构成,表示当前文件的第一个索引条目的起始位偏移量。与mappedFile文件名不同的是,其后续文件名是固定的。因为consumequeue文件大小是固定不变的

索引条目

每个consumequeue文件可以包含30w个索引条目,每个索引条目包含了三个消息重要属性:

  1. 消息在mappedFile文件中的偏移量CommitLog Offset,8字节
  2. 消息长度,4字节
  3. 消息Tag的hashcode值,8字节

这三个属性占20个字节,所以每个文件的大小固定是30w*20字节

对文件的读写

消息写入:

  1. broker根据生产者发送过来的queueId,获取到该消息对应的索引条目要在consumequeue中的写入偏移量,即QueueOffset
  2. 将queueId、queueOffset等数据,与消息一起封装为消息单元
  3. 将消息单元写入到commitlog。同时,形成消息索引条目
  4. 将消息索引条目分发到相应的consumequeue

消息拉取:

  1. consumer获取到其要消费消息所在queue的消费offset(消费进度),计算出其要消费消息的消息offset(消费offset+1)
  2. consumer向broker发送拉取请求,其中会包含其要拉取消息的queue、消息offset及消息tag
  3. broker计算在该consumequeue中的queueOffset(queueOffset=消息offset*20字节)
  4. 从该queueOffset处开始向后查找第一个指定tag的索引目录
  5. 解析该索引目录的前8个字节,即可定位到该消息在commitlog中的commitlog offset
  6. 从对应commitlog offset中读取消息单元,并发送给consumer

性能优化

  • rocketmq对文件的读写操作是通过mmap零拷贝进行的,将对文件的操作转化为直接对内存地址进行操作,从而极大的提高了文件的读写效率
  • consumequeue中的数据是顺序存放的,还引入了PageCache预读取机制,使得对consumequeue文件的读取几乎接近于内存读取,即使在有消息堆积情况下也不会影响性能

PageCache机制:

  • 写操作:os会先将数据写入到PageCache中,随后会以异步的方式由pdflush内核线程将cache中的数据刷盘到物理磁盘
  • 读数据:先从PageCache中读取数据,若没有命中,则os再从物理磁盘上加载数据到PageCache,同时也会对相邻数据块中的数据进行预读取

rockermq中可能会影响性能的是对commitlog的读取。因为对commitlog来说,读取消息时会产生大量的随机访问

indexFile

rocketmq提供了根据key进行消息查询的功能。该查询是通过store目录中的index子目录中的indexFile进行索引实现的快速查询。当然,这个indexFile中的索引数据是包含了key的消息被发送到了broker时写入的,如果消息中没有包含key,则不会写入

indexFile的创建时机:

  • 当一条条带key的消息发送过来后,发现没有indexFile,此时会创建
  • 当一个indexFile中挂载的index索引单元数量超出2000w个时,会创建新的indexFile(根据indexHeader中的indexCount)

indexFile结构

每个broker中会包含一组indexFile,每个indexFile都是以一个文件创建的时间戳命名的。每个indexFile文件由三部分构成:indexHeader、slots槽位、indexes索引单元。每个indexFile文件中包含500w个slot槽。而每个slot槽又可能会挂载很多的index索引单元

一个indexFile的最大大小为:40+500w*4+2000w*20个字节

indexFile的文件名为当前被创建的时间戳,根据业务key进行查询时,查询条件除了key之外,还需要额外指定一个要查询的时间戳,表示要查询不大于该时间戳的最新消息,这个时间戳文件名可以简化查询,提高效率

indexHeader

indexHeader固定40个字节

  • beginTimestamp:该indexFile中第一条消息的存储时间
  • endTimestamp:该indexFile中最后一条消息存储时间
  • beginPhyoffset:该indexFile中第一条消息在commitlog中的偏移量commitlog offset
  • endPhyoffset:该indexFile中最后一条消息在commitlog中的偏移量commitlog offset
  • hashSlorCount:已经填充有index的slot数量
  • indexCount:该indexFile中包含的索引个数

slots槽位

key的hash值%500w的结果即为slot槽位,然后将该slot值修改为该index索引单元的indexNo,根据这个indexNo可以计算出该index单元在indexFile中的位置。不过,该取模结果的重复率是很高的,为了解决这个问题,在每个index索引单元中添加了preIndexNo,用于指定该slot中当前index索引单元的前一个index索引单元。而slot中始终存放的是其下最新的index索引单元的indexNo,这样的话,只要找到了slot就可以找到最新的index索引单元,而通过这个index索引单元就可以找到其之前的所有index索引单元

indexNo是一个在indexFile中的流水号,从0开始递增

index索引单元

index索引单元默认20个字节,包含4个属性

  • keyHash:消息中指定的业务key的hash值
  • phyOffset:当前key对应的消息在commitlog中的偏移量commitlog offset
  • timeDiff:当前key对应消息的存储时间与当前indexFile创建时间的时间差
  • preIndexNo:当前slot下当前index索引单元的前一个索引单元的indexNo

indexNo在index索引单元中是没有体现的,其是通过indexes中依次数出来的

查询流程

定位公式:

  • 计算指定消息key的slot槽位序号:key的hash%500w
  • 计算槽位为n的slot在indexFile中的起始位置:40+(n-1)*4
  • 计算indexNo为m的index在indexFile中的位置:40+500w*4+(m-1)*20
  1. 输入业务key与要查询的时间,开始查询
  2. 根据传入的时间找到相应的indexFile(文件名<=指定查询时间 的最大的文件)
  3. 计算出传入时间与找到的indexFile文件名间的差值diff
  4. 计算出业务key的hash值
  5. 计算出slot槽位序号(key的hash%500w)
  6. 根据slot槽位序号计算出该slot在indexFile中的位置(40+(n-1)*4)
  7. 找到slot后读取slot值,即当前slot中最新的index索引单元的indexNo
  8. 根据indexNo计算出该index单元在indexFile的位置(40+500w*4+(m-1)*20)
  9. 之前计算的时间差diff减去当前index单元中的timeDiff
    • 如果>=0:读取该index单元的phyOffset,然后就可以定位到commitlog中的消息了
    • 如果<0:读取该index单元的preIndexNo,作为要查找的下一个index索引单元的indexNo(越向前,index单元中的timeDiff越小)

消息的消费

消费者从broker中获取消息的方式有两种:pull拉取和push推送。消费者组对消息的消费模式又分为两种:集群消费clustering和广播消费broadcasting

获取消息类型

  • 拉取模式:consumer主动从broker中拉取消息,主动权由consumer控制。这种方式实时性较弱

  • 推送模式:broker收到数据后会主动推送给consumer。该模式实时性较高

    该消费类型是典型的发布-订阅模式,即consumer向其关联的queue注册监听器,一旦有新消息就会触发回调,回调方式是consumer去queue中拉取消息。而这需要基于consumer和broker间的长连接。长连接是需要消耗系统资源的

对比:

  • pull:需要应用去实现对关联queue的遍历,实时性差;但便于应用控制消息的拉取
  • push:封装了对关联queue的遍历,实时性强,但会占用较多系统资源

组消费模式

  • 广播模式:相同consumer group中的每个consumer实例都接收同一个topic的全量消息。即每条消息都会被发送到consumer group中的每个consumer
  • 集群模式:相同consumer group中的每个consumer实例平均分摊同一个topic的消息。即每条消息只会被发送到consumer group中的某个consumer

消息进度保存

  • 广播模式:消费进度保存在consumer端。因为广播模式下consumer group中每个consumer都会消费所有消息,但他们的消费进度是不同的。所以consumer各自保存各自的消费进度
  • 集群模式:消费进度保存在broker中。consumer group中的所有consumer共同消费同一个topic中的消息,同一条消息只会被消费一次。消费进度参与到了消费的负载均衡中,故消费进度是需要共享的(config/consumerOffset.json中)

rebalance机制

rebalance即再均衡,是指将一个topic下的多个queue在同一个consumer group中的多个consumer间进行重新分配的过程

集群模式才会有rebalance机制

rebalance机制是为了提升消息的并行消费能力。例如:一个topic下5个队列,在只有一个消费者的情况下,这个消费者要负责这五个队列的消息。如果此时增加一个消费者,那么就可以给其中一个消费者分配两个,另一个分配三个,从而提升并行消费能力

限制

由于一个队列最多分配给一个消费者,因此当某个消费者组下的消费者实例数量大于队列数量时,多余的消费者将分配不到任何队列

危害

  • 消费暂停:在触发rebalance时consumer要暂停部分队列的消费,要等rebalance结束时才能继续消费

  • 消费重复:consumer在消费新分配给自己的队列时,必须接着之前consumer提交的消费进度的offset继续消费。然而默认情况下,offset是异步提交的,可能会导致提交的offset和实际的offset不符,导致重复消费

    • 同步提交:提交offset后,需要等待broker返回ack后才能获取下一批消息

    • 异步提交:提交offset后,不需要等待broker的ack就可以直接获取下一批消息

  • 消费突刺:由于rebalance可能导致重复消费,如果需要重复消费的消息过多,或者因为rebalance暂停时间过长从而导致积压了部分消息。那么有可能会导致rebalance结束后瞬间需要消费很多消息

原因

导致rebalance产生的原因:

  • 消费者所订阅的topic的queue数量发生变化
  • 消费者组中消费者数量发生变化

过程

  1. 在broker中维护着多个map集合,这些集合中动态存放着当前topic中的queue信息、consumer group中的consumer实例信息。一旦发现消费者所订阅的queue数量发生变化,或消费者组中消费者的数量发生变化,立即向consumer group中的每个实例发出rebalance通知
  2. consumer实例接收到通知后会采用queue分配算法自己获取到相应的queue,即由consumer实例自主进行rebalance

在kafka中,一旦发现出现了rebalance条件,broker会调用group coordinator来完成rebalance。group coordinator是broker中的一个进程。

group coordinator会在consumer group中选出一个group leader。由这个leader根据自己本身情况完成partition分区的再分配。这个再分配结果会上报给coordinator,并由coordinator同步给group中的所有consumer实例

kafka中的的rebalance是由consumer leader完成的。而rockermq是由每个consumer自身完成的

queue分配算法

  • 平均分配策略:该算法是要根据avg = QueueCount / ConsumerCount的结果进行分配的。如果能够整除,则按顺序将avg个queue逐个分配给Consumer;如果不能整除,则将多余的queue按照consumer顺序逐个分配
  • 环形分配策略:根据消费者的顺序,依次在由queue队列组成的环形图中逐个分配
  • 一致性hash策略:将consumer的hash值作为node节点存放到hash环上,然后将queue的hash值也放到hash环上,通过顺时针的方向,距离queue最近的那个consumer就是该queue要分配的consumer
  • 同机房策略:根据queue的部署机房位置和consumer的位置,过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue,则按照平均分配策略或环形分配策略对所有queue进行分配

至少一次原则

rocketm有一个原则:每条消息必须要被成功消费一次

consumer在消费完消息后会向其消费进度记录器提交其消费消息的offset,offset被成功记录到记录器中,那么这条消息就被成功消费了

消费进度记录器:

  • 广播模式:consumer本身就是消费进度记录器
  • 集群模式:broker是消费进度记录器

offset管理

消费进度offset是用来记录每个queue在不同消费者组的消费进度。根据消费进度记录器的不同,可以分为两种模式:本地模式和远程模式

本地模式

当消费者模式为广播模式时,offset使用本地模式存储。因为每条消息都会被所有消费者消费,每个消费者管理自己的消费进度,各个消费者直接不存在消费进度的交集

consumer在广播模式下offset相关数据以json的形式持久化到consumer本地磁盘文件中,默认文件路径为当前用户目录下的.rocketmq_offsets/${clinetId}/${group}/Offsets.json。其中${clientId}为当前消费者id,默认为ip@DEFAULT;${group}为消费者组名

远程模式

当消费模式为集群模式时,offset使用远程模式管理。因为所有consumer实例对消息采用的是均衡消费,所有consumer共享queue的消费进度

consumer在集群模式下offset相关数据以json的形式持久化到broker磁盘中,文件路径为当前用户主目录下的store/config/consumerOffset.json

broker启动时会加载这个文件,并写入到一个双层map。外层map的key为topic@group,value为内层map。内层map的key为queueId,value为offset。当发生rebalance时,新的consumer会从该map中获取到相应的数据来继续消费

使用远程模式不光是因为共享消费进度,也是为了可以方便进行rebalance

用途

消费者要消费的第一条消息的起始位置是用户通过consumer.setConsumeFromWhere()方法指定的

在consumer启动后,要消费的第一条消息的起始位置常用的三种(ConsumeFromWhere枚举):

  • CONSUME_FROM_LAST_OFFSET:从queue的的当前最后一条消息开始消费
  • CONSUME_FROM_FLRST_OFFSET:从queue的第一条消息开始消费
  • CONSUME_FROM_TIMESTAMP:从指定的具体时间戳位置的消息开始消费。时间戳是通过另外一个语句指定的cosumer.setConsumerTimestamp(“yyyyMMddHHmmss”)

当消费完一批消息后,consumer会提交其消费进度offset给broker,broker在收到消费进度后会将其更新到哪个双层map及consumerOffset.json文件中,然后向该consumer进行ack,而ack内容包含:当前消费队列的最小offset(minOffset)、最大offset(maxOffset)、及下次消费的起始offset(nextBeginOffset)

重试队列

当rocketmq对消息的消费出现异常时,会将发生异常的消息的offset提交到broker中的重试队列。系统在发生消息消费异常时会为当前topic@group创建一个重试队列,该队列以%RETRY%开头,到达重试时间后进行消费重试

同步提交和异步提交

集群模式下,consumer消费完消息后会向broker提交消费进度offset,其提交方式分为:

  • 同步提交:消费者在消费完一批消息后会向broker提交这些消息的offset,然后等待broker的成功响应。若在等待超时之前收到了成功响应,则继续读取下一批消息进行消费(从ack中获取nextBeginOffset)。若没有收到响应,则会重新提交,直到获取到响应。而在这个等待的过程中,消费者是阻塞的。其严重影响了消费者的吞吐量
  • 异步提交:消费者在消费完一批消息后向broker提交offset,但无需等待broker的成功响应,可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意,broker在收到提交的offset后,还是会向消费者进行响应的。在没有收到ack时,consumer会直接从broker中获取nextBeginOffset

消费幂等

当出现消费者对某条消息重复消费时,重复消费的结果与消费一次的结果是相同的,并且多次消费并未对业务系统产生任何负面影响,那么这个消费过程就是幂等的

幂等解决方案的设计中涉及到两项要素:幂等令牌,与唯一性处理。只要充分利用好这两项要素,就可以设计好的幂等解决方案

  • 幂等令牌:是生产者和消费者两者间的既定协议,通常指具备唯一业务标识的字符串
  • 唯一性处理:服务端通过采用一定的算法策略,保证同一个业务逻辑不会被重复执行成功多次

解决方案

  1. 首先通过缓存去重。在缓存中如果已经存在了幂等令牌,则说明本次操作是重复性操作;若缓存没有命中,则进入下一步
  2. 在唯一性处理之前,先在数据库中查询幂等令牌作为索引的数据是否存在。若存在,则说明本次操作为重复性操作;若不存在,则进入下一步
  3. 唯一性处理后,将幂等令牌写入到缓存,并将幂等令牌作为唯一索引写入到db
  • 1中的缓存过期是有可能直接进入2中的,1可以帮2减轻压力
  • rocketmq中不要用MessageID作为幂等令牌,因为会重复
  • rocketmq本身可以保证消息不丢失,但不能保证消息不重复

消息堆积和消费延迟

消息处理流程中,如果consumer的消费速度跟不上producer的发送速度,mq中的未处理消息会越来越多,这部分消息就称为“堆积消息”。消息堆积进而会造成“消费延迟”

consumer使用长轮询pull模式消费消息时,分为两个阶段:

  1. 拉取消息:consumer通过长轮询pull模式批量拉取的方式从服务端获取消息,将拉取到的消息缓存到本地缓冲队列中。对于拉取式消费,在内网环境下会有很高的吞吐量,所以这一阶段不会成为消息堆积的瓶颈
  2. 消费消息:consumer将本地缓存的消息提交到消费线程中,使用业务消费逻辑对消费进行处理,处理完毕后获取到一个结果。这是真正的消息消费过程。此时consumer的消费能力就完全依赖于消息的消费耗时和消费并发度了。如果因为业务逻辑复杂,导致单条消息的耗时过长,则整体的吞吐量肯定不高,此时会导致consumer本地缓冲队列达到上限,停止从服务拉取消息

消息堆积的主要瓶颈在于客户端的消费能力,而消费能力由消费耗时和消费并发度决定(消费耗时的优先级高于消费并发度)

普通消息、延时消息、事务消息:消费并发度=单节点线程数*节点数量

顺序消息:

  • 全局顺序消息:这类topic只有一个queue,consumer group中只有一个consumer能消费,并发度为1
  • 分区顺序消息:这类topic有多个queue,只能保证单个queue的顺序消费。消息并发度=topic的queue分区数量

消息的清理

消息被消费过后不会被清理掉

消息是被顺序存储在commitlog文件的,且消息大小不定长,所以消息的清理是不可能以消息为单位进行清理的,而是以commitlog文件为单位进行清理的。否则会降低清理效率,并且实现逻辑复杂

commitlog文件存在一个过期时间,默认为72小时,即三天。除了用户手动清理外,在以下情况也会被自动清理(无论文件中的消息是否被消费过):

  • 文件过期,且到达清理时间点(默认凌晨4点)后,自动清理过期文件
  • 文件过期,且磁盘空间占用率已达到过期清理警戒线(默认75%)后,无论是否达到清理时间点,都会自动清理过期文件
  • 磁盘占用率达到清理警戒线(默认85%)后,开始按照设定好的规则清理文件,无论是否过期。默认从最老的文件开始清理
  • 磁盘占用率达到危险警戒线(默认90%)后,broker将拒绝消息写入

官方建议rocketmq服务的Linux文件系统采用ext4。因为对于文件删除操作,ext4比ext3性能更好