目录

rocketmq基本概念介绍

部署架构

http://img.cana.space/picStore/20201124224026.png

角色介绍

  • Producer:消息的发送者;举例:发信者
  • Consumer:消息接收者;举例:收信者
  • Broker:暂存和传输消息;举例:邮局
  • NameServer:管理Broker;相当于注册中心,保存了broker的路由信息
  • Topic:区分消息的种类;一个发送者可以发送消息给一个或者多个Topic;一个消息的接收者可以订阅一个或者多个Topic消息
  • Message Queue:相当于是Topic的分区;用于并行发送和接收消息

Message

  • topic:主题名称
  • tag:消息TAG,用于消息过滤对消息的整体分类,比如 topic为物流跟踪轨迹 ,轨迹包含 揽收 出库 入库 派送 签收,可以分别给这些相同topic不同类型的数据打标签分类解析处理
  • keys:Message索引键,多个用空格隔开,RocketMQ可以根据这些key快速检索到消息对消息关键字的提取方便查询,比如一条消息某个关键字是 运单号,之后我们可以使用这个运单号作为关键字进行查询
  • waitStoreMsgOK:消息发送时是否等消息存储完成后再返
  • delayTimeLevel:消息延迟级别,用于定时消息或消息重
  • User property:自定义消息属性

消息查询介绍

RocketMQ提供了3种消息查询方式:

  • **按照Message Key 查询:**消息的key是业务开发同学在发送消息之前自行指定的,通常会把具有业务含义,区分度高的字段作为消息的key,如用户id,订单id等。
  • **按照Unique Key查询:**除了业务开发同学明确的指定消息中的key,RocketMQ生产者客户端在发送发送消息之前,会自动生成一个UNIQ_KEY,设置到消息的属性中,从逻辑上唯一代表一条消息。
  • **按照Message Id 查询:**Message Id 是消息发送后,在Broker端生成的,其包含了Broker的地址,和在CommitLog中的偏移信息,并会将Message Id作为发送结果的一部分进行返回。Message Id中属于精确匹配,从物理上唯一代表一条消息,查询效率更高。

RocketMQ有意弱化Unique Key与Message Id的区别,有时都称之为Message Id。在通过RocketMQ的命令行工具或管理平台进行查询时,二者可以通用。在根据Unique Key进行查询时,本身是有可能查询到多条消息的,但是查询工具会进行过滤,只会返回一条消息。这种情况导致很多RocketMQ的用户,并未能很好对二者进行区分。

业务开发同学在使用RocketMQ时,应该养成良好的习惯,在发送/消费消息时,将这些信息记录下来,通常是记录到日志文件中,以便在出现问题时进行排查。

事实上,用户主动设置的Key以及客户端自动生成的Unique Key,最终都会设置到Message对象的properties属性中,如下图所示:

http://img.cana.space/picStore/20201129005048.png

其中:

  • **KEYS:**表示用户通过setKeys方法设置的消息key,
  • **UNIQ_KEY:**表示消息发送之前由RocketMQ客户端自动生成的Unique Key。细心的读者发现了其值与上述打印SendResult结果中的msgId字段的值是一样的,这验证了前面所说的msgId表示的实际上就是Unique Key的说法。

NameServer集群

NameServer的作用是注册中心,类似于Zookeeper,但又有区别于它的地方。每个NameServer节点互相之间是独立的,没有任何信息交互,也就不存在任何的选主或者主从切换之类的问题,因此NameServer与Zookeeper相比更轻量级。单个NameServer节点中存储了活跃的Broker列表(包括master和slave),这里活跃的定义是与NameServer保持有心跳。

Broker集群

Broker是具体提供业务的服务器,单个Broker节点与所有的NameServer节点保持长连接及心跳,并会定时将Topic信息注册到NameServer,顺带一提底层的通信和连接都是基于Netty实现的。 Broker中分master和slave两种角色,每个master可以对应多个slave,但一个slave只能对应一个master,master和slave通过指定相同的Brokername,不同的BrokerId (master为0)成为一个组。master和slave之间的同步方式分为同步双写和异步复制,异步复制方式master和slave之间虽然会存在少量的延迟,但性能较同步双写方式要高出10%左右。

另外,Broker中还存在一些非常重要的名词需要说明:

rocketmq中的topic、topic分片和queue

RocketMQ的Topic/Queue和JMS中的Topic/Queue概念有一定的差异,JMS中所有消费者都会消费一个Topic消息的副本,而Queue中消息只会被一个消费者消费;但到了RocketMQ中Topic只代表普通的消息队列,而Queue是组成Topic的更小单元,集群消费模式下一个消费者只消费该Topic中部分Queue中的消息,当一个消费者开启广播模式时则会消费该Topic下所有Queue中的消息。Topic和Queue的具体关系可以参考下图

http://img.cana.space/picStore/20201124224225.png

从本质上来说,RocketMQ中的Queue是数据分片的产物。

为了更好地理解Queue的定义,我们还需要引入一个新的概念:Topic分片。在分布式数据库和分布式缓存领域,分片概念已经有了清晰的定义。同理,对于RocketMQ,一个Topic可以分布在各个Broker上,我们可以把一个Topic分布在一个Broker上的子集定义为一个Topic分片。对应上图,TopicA有3个Topic分片,分布在Broker1,Broker2和Broker3上,TopicB有2个Topic分片,分布在Broker1和Broker2上,TopicC有2个Topic分片,分布在Broker2和Broker3上。

将Topic分片再切分为若干等分,其中的一份就是一个Queue。每个Topic分片等分的Queue的数量可以不同,由用户在创建Topic时指定。

我们知道,数据分片的主要目的是突破单点的资源(网络带宽,CPU,内存或文件存储)限制从而实现水平扩展。RocketMQ 在进行Topic分片以后,已经达到水平扩展的目的了,为什么还需要进一步切分为Queue呢?

解答这个问题还需要从负载均衡说起。以消息消费为例,借用Rocket MQ官方文档中的Consumer负载均衡示意图来说明:

如图所示,TOPIC_A在一个Broker上的Topic分片有5个Queue,一个Consumer Group内有2个Consumer按照集群消费的方式消费消息,按照平均分配策略进行负载均衡得到的结果是:第一个 Consumer 消费3个Queue,第二个Consumer 消费2个Queue。如果增加Consumer,每个Consumer分配到的Queue会相应减少。Rocket MQ的负载均衡策略规定:Consumer数量应该小于等于Queue数量,如果Consumer超过Queue数量,那么多余的Consumer 将不能消费消息。

在一个Consumer Group内,Queue和Consumer之间的对应关系是一对多的关系:一个Queue最多只能分配给一个Consumer,一个Cosumer可以分配得到多个Queue。这样的分配规则,每个Queue只有一个消费者,可以避免消费过程中的多线程处理和资源锁定,有效提高各Consumer消费的并行度和处理效率。

由此,我们可以给出Queue的定义:

Queue是Topic在一个Broker上的分片等分为指定份数后的其中一份,是负载均衡过程中资源分配的基本单元。

Tags

Tags是Topic下的次级消息类型(注:Tags也支持TagA || TagB这样的表达式),可以在同一个Topic下基于Tags进行消息过滤。Tags的过滤需要经过两次比对,首先会在Broker端通过Tag hashcode进行一次比对过滤,匹配成功传到consumer端后再对具体Tags进行比对,以防止Tag hashcode重复的情况。Queue中具体的存储单元结构如下图:

http://img.cana.space/picStore/20201124230935.png

Producer集群

与nameserver的关系

单个Producer和一台nameserver保持长连接,定时查询topic配置信息,如果该nameserver挂掉,生产者会自动连接下一个nameserver,直到有可用连接为止,并能自动重连。与nameserver之间没有心跳

与broker的关系

单个Producer和与其关联的所有broker保持长连接,并维持心跳。默认情况下消息发送采用轮询方式,会均匀发到对应Topic的所有queue中。

最佳实践

  • 一个应用尽可能只使用一个 Topic,消息子类型用 tags 来标识,tags 可以由应用自由设置。只有发送消息设置了tags,消费方在订阅消息时,才可以利用 tags 在 broker 做消息过滤。
  • 每个消息在业务层面的唯一标识码,要设置到 keys 字段,方便将来定位消息丢失问题。服务器会为每个消 息创建索引(哈希索引),应用可以通过 Topic,key 来查询返条消息内容,以及消息被谁消费。由于是哈希索引,请务必保证 key 尽可能唯一,这样可以避免潜在的哈希冲突。
  • 消息发送成功或者失败,要打印消息日志,务必要打印 sendresult 和 key 字段。
  • 对于消息不可丢失应用,务必要有消息重发机制。例如:消息发送失败,存储到数据库,能有定时程序尝试重发或者人工触发重发。
  • 某些应用如果不关注消息是否发送成功,请直接使用sendOneWay方法发送消息。

Consumer集群

与nameserver的关系

单个Consumer和一台nameserver保持长连接,定时查询topic配置信息,如果该nameserver挂掉,消费者会自动连接下一个nameserver,直到有可用连接为止,并能自动重连。与nameserver之间没有心跳。

与broker的关系

单个Consumer和与其关联的所有broker保持长连接,并维持心跳,失去心跳后,则关闭连接,并向该消费者分组的所有消费者发出通知,分组内消费者重新分配队列继续消费。

最佳实践

  • Consumer 数量要小于等于queue的总数量,由于Topic下的queue会被相对均匀的分配给Consumer,如果 Consumer 超过queue的数量,那多余的 Consumer 将没有queue可以消费消息。
  • 消费过程要做到幂等(即消费端去重),RocketMQ为了保证性能并不支持严格的消息去重。
  • 尽量使用批量方式消费,RocketMQ消费端采用pull方式拉取消息,通过consumeMessageBatchMaxSize参数可以增加单次拉取的消息数量,可以很大程度上提高消费吞吐量。另外,提高消费并行度也可以通过增加Consumer处理线程的方式,对应参数consumeThreadMinconsumeThreadMax
  • 消息处理成功或者失败,要打印消息日志。

相关总结

RocketMQ如何做到高吞吐、低延迟

消息发送的高性能

  • 顺序写
  • 定长,CommitLog 1G, ConsumeQueue 30w个条目 x 20字节,内存映射可以映射一段固定长度
  • 消息发送,使用异步刷盘

MQClientInstance什么时候会复用

复用这种情况出现在你在一个JVM里启动了多个Producer时,且没有设置instanceName和unitName,那么这两个Producer会公用一个MQClientInstance,发送的消息会路由到同一个集群。例如,你起了两个Producer,并且配置的NameServer地址不一样,本意是让这两个Producer往不同集群上分配消息,但是由于共用了一个MQClientInstance,这个MQClientInstance是基于先来的Producer配置构建的,第二个Producer和他公用后被认为是同一instance,配置是相同的,消息的路由就是相同的,就没有达到你想要的效果。