当前位置：首页 > news >正文

中间件-RocketMQ

news 来源：原创 2025/8/15 11:55:05

RocketMQ

基本架构
消息模型
消费者消费消息模式
顺序消息机制
延迟消息
批量消息
事务消息
消息重试
最佳实践

基本架构

在这里插入图片描述

nameServer: 维护broker列表信息，客户端连接时只需要连接nameServer。可配置成集群。
broker：broker分为master和slave，master负责消息的发送和消费，slave是master的备份。master-slaver的集群方式，master挂掉时候slave不能主动转换为master提供服务（5.X版本后可以通过配置实现mater挂掉后slave转为master提供服务）。
leader-follower的集群方式，即高可用集群，各个broker是对等的，通过选举产生leader（在dashboart中显示为master），如果leader挂掉，在剩下的follower（显示为slave）中选举再产生新的leader。注意，只有超过半数的几点存活，才能选举出leader。

消息模型

在这里插入图片描述
⽣产者和消费者都可以指定⼀个Topic发送消息或者拉取消息。⽽Topic是⼀个逻辑概念。
Topic中的消息会分布在后⾯多个MessageQueue当中。这些MessageQueue会分布到⼀个或者多个broker中。

消费者消费消息模式

广播模式：所有关注topic的消费者都收到消息。广播模式下消息队列的消费位点由客户端自己维护，消费失败服务端不会重发。
集群模式：同一个消费者组只有一个成员收到消息。集群模式下消费点位由服务端维护，消费者组的所有成员共用一个位点，消费失败服务端会重发。

顺序消息机制

⽣产者只有将⼀批有顺序要求的消息，放到同⼀个MesasgeQueue上，通过MessageQueue的FIFO特性保证这⼀批消息的顺序。如果不指定MessageSelector对象，
那么⽣产者会采⽤轮询的⽅式将多条消息依次发送到不同的MessageQueue上。
消费者需要实现MessageListenerOrderly接⼝，实际上在服务端，处理MessageListenerOrderly时，会给⼀个MessageQueue加锁，拿到MessageQueue上所有的消息，然后再去读取下⼀个MessageQueue的消息。
消费消息失败时，不建议抛出异常，可以返回ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT作为替代。因为消费者端只进⾏有限次数的重试。如果⼀条消息处理失败，RocketMQ会将后续消息阻塞住，让消费者进⾏重试。但是，如果消费者⼀直处理失败，超过最⼤重试次数，那么RocketMQ就会跳过这⼀条消息，处理后⾯的消息，这会造成消息乱序。

延迟消息

定固定的延迟级别：对于指定固定延迟级别的延迟消息，RocketMQ的实现⽅式是预设⼀个系统Topic，名字叫做SCHEDULE_TOPIC_XXXXX。在这个Topic下，预设了18个MessageQueue。这⾥每个对列就对应了⼀种延迟级别。然后每次扫描这18个队列⾥的消息，进⾏延迟操作就可以了。
指定消息发送时间：RocketMQ是通过时间轮算法实现。

批量消息

⽣产者要发送的消息⽐较多时，可以将多条消息合并成⼀个批量消息，⼀次性发送出去。这样可以减少⽹络IO，提升消息发送的吞吐量。同⼀批消息的Topic必须相同，另外，不⽀持延迟消息。还有批量消息的⼤⼩不要超过1M，如果太⼤就需要⾃⾏分割。

事务消息