• 业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
• 消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。
• 消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。
总的来说，针对消息分类，您可以选择创建多个 Topic，或者在同一个 Topic 下创建多个 Tag。但通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息，例如全集和子集的关系、流程先后的关系。
通过合理的使用 Topic 和 Tag，可以让业务结构清晰，更可以提高效率。
RocketMQ分布式消息队列的消息过滤方式有别于其它MQ中间件，是在Consumer端订阅消息时再做消息过滤的。RocketMQ这么做是在于其Producer端写入消息和Consumer端订阅消息采用分离存储的机制来实现的，Consumer端订阅消息是需要通过ConsumeQueue这个消息消费的逻辑队列拿到一个索引，然后再从CommitLog里面读取真正的消息实体内容，所以说到底也是还绕不开其存储结构。其ConsumeQueue的存储结构如下，可以看到其中有8个字节存储的Message Tag的哈希值，基于Tag的消息过滤正式基于这个字段值的。
Consumer端在订阅消息时除了指定Topic还可以指定TAG，如果一个消息有多个TAG，可以用||分隔。其中，Consumer端会将这个订阅请求构建成一个 SubscriptionData，发送一个Pull消息的请求给Broker端。Broker端从RocketMQ的文件存储层—Store读取数据之前，会用这些数据先构建一个MessageFilter，然后传给Store。Store从 ConsumeQueue读取到一条记录后，会用它记录的消息tag hash值去做过滤，由于在服务端只是根据hashcode进行判断，无法精确对tag原始字符串进行过滤，故在消息消费端拉取到消息后，还需要对消息的原始tag字符串进行比对，如果不同，则丢弃该消息，不进行消息消费。
RocketMQ是一款分布式、队列模型的消息中间件，单机支持1万以上的持久化队列，前提是足够的内存、硬盘空间。
消息队列主要的应用场景：异步处理，应用解耦，流量削峰，消息通讯。
如上图所示， RocketMQ的部署结构有以下特点：
Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有Name Server。
Producer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
Consumer与Name Server集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。
消息的生产者，负责发送消息，将消息推送给broker。一般由业务系统负责产生消息。
消息有3种发送方式：同步、异步、单向。
rocketmq的核心组件，负责消息的接收、存储（持久化到磁盘）、被消费者拉取消息等功能。
broker也存储消息相关的元数据，包括：消费者组、消费进度、topic&queue信息等。
消息的消费者，从broker上拉取消息从而进行消费。rocketmq提供两种消费者。
一般是后台系统负责异步消费消息。
主动消费者：DefaultMQ Pull Consumer，从broker中拉取一批消息并消费，主动权由消费者控制。
被动消费者：DefaultMQ Push Consumer，消费者实现回调接口，一旦有消息，broker回调接口，消费者被动响应。
注册中心的作用，提供轻量级的服务发现和提供路由信息（broker的服务注册与发现）。
nameserver存有全量的路由信息，提供对等的读写服务，支持快速扩缩容。
nameserver接收broker的请求，注册broker的路由信息。
nameserver接收client（producer/consumer）的请求，根据消息的topic获取相应的broker路由信息。
（手动创建的topic可以指定broker，自动创建的topic会随机指定broker，也许指定单个或全部，topic的概念在后面。）
一种消息的逻辑分类（消息的类型），比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类存储。
生产者方面：发消息时需指定topic，可以有1-n个生产者发布1个topic的消息，
也1个生产者可以发布不同topic的消息。消费者方面：收消息时需订阅topic，
可以有1-n个消费者组订阅1个topic的消息，1个消费者组可以订阅不同topic的消息。
1个消息必须指定1个topic，topic允许自动创建与手工创建，topic创建时需要指定broker，可以指定1个或多个，
name server就是通过broker与topic的映射关系来做路由。
producer和consumer在生产和消费消息时，都需要指定消息的 topic，当topic匹配时，
consumer 才会消费到producer发送的消息。
topic与broker是多对多的关系，一个topic分布在多个broker上，一个broker可以配置多个topic。
message是消息的载体。每个message必须指定一个topic，相当于寄信的地址。
message还有一个可选的tag设置，以便消费端可以基于tag进行过滤消息。
message还有扩展的kv结构，例如你可以设置一个业务key到你的消息中，在broker上查找消息并诊断问题。
标签可以被认为是对topic的进一步细化。一般在相同业务模块中通过引入标签来标记不同用途的消息。
区分相同topic下不同种类的消息。
生产到哪个topic的哪个tag下，消费者也是从topic的哪个tag进行消费，即实现消息的过滤。
queue是消息的物理管理单位，而topic是逻辑管理单位。一个topic下可以有多个queue，
默认自动创建是4个，手动创建是8个。
queue的引入使得消息存储可以分布式集群化，具有了水平扩展的能力。
1个message只能属于1个queue、1个topic。
在rocketmq中，所有消息队列都是持久化，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，
访问其中的存储单元使用offset来访问，offset 为 java long 类型，64 位，理论上在 100年内不会溢出，
所以认为是长度无限，另外队列中只保存最近几天的数据，之前的数据会按照过期时间来删除。
也可以认为 Message Queue是一个长度无限的数组，offset就是下标。
rocketmq中，producer将消息发送给broker时，需要指定发送到哪一个queue中，默认情况下，
producer会轮询的将消息发送到每个queue中，顺序是随机的，但总体上每个queue的消息数量均分，
所有broker下的queue合并成一个list去轮询，
也可以由程序员通过MessageQueueSelector接口来指定具体发送到哪个queue中。
对于consumer而言，会为每个consumer分配固定的队列（如果队列总数没有发生变化），
consumer从固定的队列中去拉取没有消费的消息进行处理。
消费端会通过RebalanceService线程，10秒钟做一次基于topic下的所有队列负载，获取同一个Consumer Group下的所有Consumer实例数或Topic的queue的个数是否改变，通知所有Consumer实例重新做一次负载均衡算法。
理解成消费进度，可自增。
虽然每个topic下面有很多message queue，但是message queue本身并不存储消息。
真正的消息存储会写在CommitLog的文件，message queue只是存储CommitLog中对应的位置信息，
方便通过message queue找到对应存储在CommitLog的消息。 不同的topic，
message queue都是写到相同的CommitLog 文件，也就是说CommitLog完全的顺序写。
服务启动顺序：name server->broker->producer&consumer
每个broker与name server集群中的所有节点建立长连接，
定时注册topic&broker的路由信息到所有name server中。
producer与name server集群中的其中一个节点（随机选择）建立长连接，
定期从name server获取topic路由信息，并向提供topic服务的broker master建立长连接，
且定期向broker master发送心跳，produce无状态，可集群部署。
producer只能将消息发送到broker master，但是consumer则不一样。
consumer与name server集群中的其中一个节点（随机选择）建立长连接，定期从name server获取topic路由信息，
consumer同时与提供topic服务的master和slave建立长连接且定时发送心跳，
consumer既可以从broker master订阅消息，也可以从broker slave订阅消息，订阅规则由broker配置决定。
broker一旦需要横向扩展，只需要启动更多的broker即可，然后把对应的topic建上，
客户端的queue集合即会变大，并且由于每个group下面的topic的配置都是独立的，
也就说可以让broker1下面的那个topic的queue数量是4，其他broker下的topic queue数量是2，
这样broker1则得到更大的负载。
参考：
同一个消费组中，不一致的订阅行为有：
不一致的订阅行为，会导致部分消息丢失。
rocket mq的tag级别的消息过滤功能，其实现原理包括两方面：
存储模型
rocket mq为每一个topic建立索引，方便消费者按照topic进行消费，其具体实现为ConsumeQueue(消息队列)。consumeQueue的设计极具技巧性，队列中每个条目长度固定，组成为：
这种设计使得消费队列具备类似数组般随机访问的特性，提高consumeQueue的读取性能。
tag过滤
tag消息过滤时：
消息丢失
为什么同一消费组，多个消费者订阅同一topic不同的tag，会发生消息丢失？
使用集群模式消费时，同一个消费组内的多个消费者共同完成topic的多个consumeQueue消费，一个消费组只会分配到其中某几个consumeQueue，且同一时间，一个consumeQueue只会分配给一个消费组。如图：
问题的核心是：同一个tag的消息会分布在不同的consumeQueue中，而队列的消息只会被一个消费者消费，那么这个consumerQueue上，不同于消费者订阅tag的消息，将不会被消费，造成堆积。
如下，消费组 consumer-group-name 下的两个消费者 consumer 1 consumer2 分别订阅了两个不同的topic topic A 与 topic B ：
由于consumerQueue的分配是以消费组为单位的，会均匀分配给消费组下的消费者，而不会在意消费者本身订阅的是哪个主题。
topic A的consumerQueue q4 分配给了consumer2，但是consumer2订阅的是topic B，因此部分topic A的消息得不到消费。
得出结论：
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）