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

014 rocketmq角色介绍

news 来源：原创 2025/9/10 4:12:31

文章目录

NameServer
- 1 服务发现机制
- 2 为什么要使⽤NameServer
- 3 NameServer如何保证数据的最终⼀致？
- 4 特点
Broker
Producer
Consumer
Topic
Queue
Producer Group
Consumer Group
Message
Tag
Offset

同一消费者组下，队列只能由一个消费者消费
广播模式：LocalFileOffsetStore offset存在消费者
集群模式：RemoteBrokerOffsetStore offset存在消费者
/store/config/consumerOffset.json

topic+消费者组：队列：offset值

Broker集群:
Broker⽤于接收⽣产者发送消息，或者消费者消费消息的请求。⼀个Broker集群由多组Master/Slave组成，Master可写可读，Slave只可以读，Master将写⼊的数据同步给Slave。每个Broker节点，在启动时，都会遍历NameServer列表，与每个NameServer建⽴⻓连接，注册⾃⼰的信息，之后定时上报。
Producer集群:
消息的⽣产者，通过NameServer集群获得Topic的路由信息，包括Topic下⾯有哪些Queue，这些Queue分布在哪些Broker上等。Producer只会将消息发送到Master节点上。
Consumer集群:
消息的消费者，通过NameServer集群获得Topic的路由信息，连接到对应的Broker上消费消息。
注意，由于Master和Slave都可以读取消息。

NameServer

1 服务发现机制

当发出请求服务时,客户端通过注册中⼼服务知道所有的服务实例。客户端
使⽤负载均衡算法选择可⽤的服务实例中的⼀个并进⾏发送。

2 为什么要使⽤NameServer

RocketMQ设计之初时参考的另⼀款消息中间件Kafka就使⽤了Zookeeper，Zookeeper其提供了Master选举、分布式锁、数据的发布和订阅等诸多功能。

事实上，在RocketMQ的早期版本，即MetaQ 1.x和MetaQ 2.x阶段，也是依赖Zookeeper的。
但MetaQ 3.x（即RocketMQ）却去掉了ZooKeeper依赖，转⽽采⽤⾃⼰的NameServer。

RocketMQ的架构设计决定了只需要⼀个轻量级的元数据服务器就⾜够了，只需要保持最终⼀致，
⽽不需要Zookeeper这样的强⼀致性解决⽅案，不需要再依赖另⼀个中间件，从⽽减少整体维护成本。

3 NameServer如何保证数据的最终⼀致？

NameServer作为⼀个名称服务，需要提供服务注册、服务剔除、服务发现这些基本功能，但是NameServer节点之间并不通信，在某个时刻各个节点数据可能不⼀致的情况下，如何保证客户端可以最终拿到正确的数据。下⾯分别从路由注册、路由剔除，路由发现三个⻆度进⾏介绍。
路由注册

对于Zookeeper、Etcd这样强⼀致性组件，数据只要写到主节点，内部会通过状态机将数据复制到其他节点，Zookeeper使⽤的是Zab协议，etcd使⽤的是raft协议。

NameServer节点之间是互不通信的，⽆法进⾏数据复制。RocketMQ采取的策略是，在Broker节点在启动的时候，轮训NameServer列表，与每个NameServer节点建⽴⻓连接，发起注册请求。
NameServer内部会维护⼀个Broker表，⽤来动态存储Broker的信息。

同时，Broker节点为了证明⾃⼰是存活的，会将最新的信息上报给NameServer，然后每隔30秒向NameServer发送⼼跳包，⼼跳包中包含 BrokerId、Broker地址、Broker名称、Broker所属集群名称、队列和brokerIP对应关系等等，然后NameServer接收到⼼跳包后，会更新时间戳，记录这个Broker的最新存活时间。 NameServer在处理⼼跳包的时候，存在多个Broker同时操作⼀张Broker表，为了防⽌并发修改Broker表导致不安全，路由注册操作引⼊了ReadWriteLock读写锁，这个设计亮点允许多个消息⽣产者并发读，保证了消息发送时的⾼并发，但是同⼀时刻NameServer只能处理⼀个Broker⼼跳包，多个⼼跳包串⾏处理。这也是读写锁的经典使⽤场景，即读多写少。

路由剔除
正常情况下，如果Broker关闭，则会与NameServer断开⻓连接，Netty的通道关闭监听器会监听到连接断开事件，然后会将这个Broker信息剔除掉。
异常情况下，NameServer中有⼀个定时任务，每隔10秒扫描⼀下Broker表，如果某个Broker的⼼跳包最新时间戳距离当前时间超多120秒，也会判定Broker失效并将其移除。
特别的，对于⼀些⽇常运维⼯作，例如：Broker升级，RocketMQ提供了⼀种优雅剔除路由信息的⽅式。如在升级⼀个Master节点之前，可以先通过命令⾏⼯具禁⽌这个Broker的写权限，发送消息到这个Broker的请求，都会收到⼀个NO_PERMISSION响应，客户端会⾃动重试其他的Broker。当观察到这个broker没有流量后，再将这个broker移除。
路由发现
路由发现是客户端的⾏为，这⾥的客户端主要说的是⽣产者和消费者。具体来说：
⽣产者，可以发送消息到多个Topic，因此⼀般是在发送第⼀条消息时，才会根据Topic获取从NameServer获取路由信息。
对于消费者，订阅的Topic⼀般是固定的，所在在启动时就会拉取。
那么⽣产者/消费者在⼯作的过程中，如果路由信息发⽣了变化怎么处理呢？如：Broker集群新增了节点，节点宕机或者Queue的数量发⽣了变化。NameServer在路由注册或者路由剔除过程中，并不会主动推送会客户端的，这意味着，需要由客户端拉取主题的最新路由信息。
事实上，RocketMQ客户端提供了定时拉取Topic最新路由信息的机制。
DefaultMQProducer和DefaultMQConsumer底层都依赖⼀个MQClientInstance类，MQClientInstance类中有⼀个startScheduledTask⽅法，⽤于根据指定的拉取时间间隔，周期性的从NameServer并获取最新的路由表，默认是30秒(ClientConfig#pollNameServerInterval)。在拉取时，会把当前启动的Producer和Consumer需要使⽤到的Topic列表放到⼀个集合中，逐个从NameServer进⾏更新。

4 特点

相对来说，nameserver 的稳定性⾮常⾼。
1 )nameserver 互相独⽴，彼此没有通信关系，单台 nameserver 挂掉，不影响其他
2 )nameserver 不会有频繁的读写，所以性能开销⾮常⼩，稳定性很⾼。

Broker

Broker：暂存和传输消息；举例：邮局
Broker是RocketMQ的核⼼，消息存储是broker的核⼼，提供了消息的接收，存储，拉取等功能，⼀般都需要保证Broker的⾼可⽤，所以会配置Broker Slave，当Master挂掉之后，Consumer仍然可以消费Slave；

Producer

Producer发送消息时候，是否应该知道应该把消息发送到哪个broker服务器中的队列中进⾏存储？producer根本不知道把消息发送到哪个borker服务器。

Consumer

consumer和nameserver保持⻓连接，每隔30s将会从nameserver服务器中查询topic路由信息，查询到路由信息信息后，就会根据ip映射⽂件从broker中消费消息。
consumer获取路由信息后，会在本地进⾏缓存。consumer和broker保持⻓连接，每隔30s向broker发送⼼跳检测，检测broker是否处于活跃状态

Topic

Topic 是消息中间件⾥⼀个重要的概念，每⼀个 Topic 代表了⼀类消息。

Queue

RocketMQ都是磁盘消息队列的模式，对于同⼀个消费组，⼀个分区只⽀持⼀个消费者来消费消息。过少的分区，会导致消费速度⼤⼤落后于消息的⽣产速度。所以在实际⽣产环境中，⼀个Topic会设置成多分区的模式，来⽀持多个消费者，提供消费消息，进⾏并发消费。
Message Queue：Topic的分区；⽤于并⾏发送和接收消息
每个Topic分⽚等分的Queue的数量可以不同，由⽤户在创建Topic时指定。
我们知道，数据分⽚的主要⽬的是突破单点的资源（⽹络带宽，CPU，内存或⽂件存储）限制从⽽实现⽔平扩展。RocketMQ 在进⾏Topic分⽚以后，已经达到⽔平扩展的⽬的了，为什么还需要进⼀步切分为Queue呢？
消费负载均衡过程中资源分配的基本单元.
queue数量指定⽅式
1、代码指定：producer.setDefaultTopicQueueNums(8);
2、配置⽂件指定
同时设置broker服务器的配置⽂件broker.properties：defaultTopicQueueNums=16
3、mqadmin命令创建
4、rocket-console控制台指定

Producer Group

⽣产者组，简单来说就是多个发送同⼀类消息的⽣产者称之为⼀个⽣产者组。⼀个⽣产者组，代表着⼀群topic相同的Producer。即⼀个⽣产者组是同⼀类Producer的组合。

Consumer Group

消费者组，和⽣产者类似，消费同⼀类消息的多个 Consumer 实例组成⼀个消费者组。⼀个消费者组，代表着⼀群topic相同，tag相同（即逻辑相同）的Consumer。通过⼀个消费者组，则可容易的进⾏负载均衡以及容错

Message

代表⼀条消息，⼀个 Message 必须指定 Topic，相当于寄信的地址。⽤户在发送时可以设置messageKey，便于之后查询和跟踪。Message 还有⼀个可选的 Tag 设置，以便消费端可以基于 Tag 进⾏过滤消息。也可以添加额外的键值对，例如你需要⼀个业务 key 来查找 Broker 上的消息，⽅便在开发过程中诊断问题。

Tag

标签可以被认为是对 Topic 进⼀步细化。
Tag 表示消息的第⼆级类型，⽐如交易消息⼜可以分为:交易创建消息，交易完成消息等。
你可能会有这样的疑问：到底什么时候该⽤ Topic，什么时候该⽤ Tag？
建议你从以下⼏个⽅⾯进⾏判断：
1、消息类型是否⼀致：如普通消息，事务消息，定时消息，顺序消息，不同的消息类型使⽤不同的Topic，⽆法通过 Tag 进⾏区分。
2、业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使⽤不同的 Topic 进⾏区分；⽽同样是天猫交易消息，电器类订单、⼥装类订单、化妆品类订单的消息可以⽤ Tag 进⾏区分。
3、消息优先级是否⼀致：如同样是物流消息，盒⻢必须⼩时内送达，天猫超市 24 ⼩时内送达，淘宝物流则相对会会慢⼀些，不同优先级的消息⽤不同的 Topic 进⾏区分。
4、消息量级是否相当：有些业务消息虽然量⼩但是实时性要求⾼，如果跟某些万亿量级的消息使⽤同⼀个 Topic，则有可能会因为过⻓的等待时间⽽『饿死』，此时需要将不同量级的消息进⾏拆分，使⽤不同的 Topic。

Offset

Offset是指某个 Topic下的⼀条消息在某个 Message Queue⾥的位置，通过 Offset的值可以定位到这条消息，或者指示 Consumer从这条消息开始向后继续处理。
Offset主要分为本地⽂件类型和 Broker代存的类型两种。
RocketMQ集群有两种消费模式
CLUSTERING 模式，也就是同⼀个 Consumer group ⾥的多个消费者每⼈消费⼀部分，各⾃收到的消息内容不⼀样。由 Broker 端存储和控制 Offset 的值，使⽤ RemoteBrokerOffsetStore结构。
BROADCASTING模式下，每个 Consumer 都收到这个 Topic 的全部消息，各个 Consumer 间相互没有⼲扰， RocketMQ使用LocalfileOffsetStore，把 Offset存到本地 (消费者)。
DefaultMQPushConsumer类⾥有个函数⽤来设置从哪⼉开始消费消息:⽐如
setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_ FIRST_OFFSET)，这个语句设置从最⼩的 Offset开始读取。
如果从队列开始到感兴趣的消息之间有很⼤的范围，⽤ CONSUME_FROM_FIRST_OFFSET参数就不合适了，可以设置从某个时间开始消费消息，⽐如Consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP),
Consumer. setConsumeTimestamp("20200723171201”)，时间戳格式是精确到秒的
注意设置读取位置不是每次都有效，它的优先级默认在 Offset Store后⾯，⽐如在DefaultMQPushConsumer 的 BROADCASTING ⽅式下，默认是从 LocalfileOffsetStore ⾥读取某个 Topic 对应 ConsumerGroup 的 Offset，当读取不到 Offset 的时候， ConsumeFromWhere 的设置才⽣效。
⼤部分情况下这个设置在 Consumer Group初次启动时有效。如果 Consumer正常运⾏后被停⽌，然后再启动，会接着上次的 Offset开始消费， ConsumeFromWhere 的设置元效。
⽣产环境上⼀般使⽤集群模式，主要记录集群模式下offset的管理，即
RemoteBrokerOffsetStore。
RocketMQ的broker端中，offset的是以json的形式持久化到磁盘⽂件中，⽂件路径为${user.home}/store/config/consumerOffset.json。

Broker端启动后，会调⽤BrokerController.initialize()⽅法，⽅法中会对offset进⾏加载，consumerOffsetManager.load()。获取⽂件内容后，序列化为ConsumerOffsetManager对象，实质是其属性ConcurrentMap<String，ConcurrentMap<Integer, Long>> offsetTable，offsetTable的数据结构为ConcurrentMap，是⼀个线程安全的容器，key的形式为topic@group（每个topic下不同消费组的消费进度），value也是⼀个ConcurrentMap，key为queueId，value为消费位移（这⾥不是offset⽽是位移）。通过对全局ConsumerOffsetManager对象就可以对各个topic下不同消费组的消费位移进⾏获取与管理。