基于Topic的消息发布与消费模式

闲话

朋友们，好久不见，不知道你们最近怎样，但相信你们一定都挺好。已经有一段时间没有更新了，个中原因不好细说，但是归根结底也许是自己懒。这个不好，大家不要学。今天主要就是想分享一下关于消息处理机制的一些想法。24m28资讯网——每日最新资讯28at.com

基本概念

24m28资讯网——每日最新资讯28at.com

1.Topic

同一个topic下消息的格式一致，例如topic为order-update-message消息的格式都是一个统一的OrderUpdateMessage的结构24m28资讯网——每日最新资讯28at.com

2.key主键

同一主键下的消息列表具有顺序性，例如key为订单号order-0001的消息列表(Queue)下，可能包含的消息列表(Queue)如下：24m28资讯网——每日最新资讯28at.com

OrderUpdateMessage(id="msg-0001", orderId = "order-0001"， action="create", ...)OrderUpdateMessage(id="msg-0002", orderId = "order-0001"， action="paid", ...)OrderUpdateMessage(id="msg-0003", orderId = "order-0001"， action="delivering", ...)OrderUpdateMessage(id="msg-0004", orderId = "order-0001"， action="modifying", ...)OrderUpdateMessage(id="msg-0005", orderId = "order-0001"， action="delivered", ...)

3.Group消费者组

同一个topic下同一个group下的消费者，对这个group下的消息队列进行抢占式消费。例如同一个消费者组group-1下的消费者consumer-1和消费者consumer-2，以及另外一个消费者组group-2下的消费者consumer-3，消息消费的结果可能如下：24m28资讯网——每日最新资讯28at.com

// consumer-1消费的消息OrderUpdateMessage(id="msg-0001", orderId = "order-0001"， action="create", ...)OrderUpdateMessage(id="msg-0002", orderId = "order-0001"， action="paid", ...)// consumer-2消费的消息OrderUpdateMessage(id="msg-0003", orderId = "order-0001"， action="modified", ...)OrderUpdateMessage(id="msg-0004", orderId = "order-0001"， action="delivering", ...)OrderUpdateMessage(id="msg-0005", orderId = "order-0001"， action="delivered", ...)// consumer-3消费的消息OrderUpdateMessage(id="msg-0001", orderId = "order-0001"， action="create", ...)OrderUpdateMessage(id="msg-0002", orderId = "order-0001"， action="paid", ...)OrderUpdateMessage(id="msg-0003", orderId = "order-0001"， action="modified", ...)OrderUpdateMessage(id="msg-0004", orderId = "order-0001"， action="delivering", ...)OrderUpdateMessage(id="msg-0005", orderId = "order-0001"， action="delivered", ...)

Kafka的消息处理机制就是以这样的形式实现的。24m28资讯网——每日最新资讯28at.com

4.优势 

生产者和消费者完全解耦，生产者无需关注是否有消费者在消费，消费者也无需知道生产者是否在生成新的消息。24m28资讯网——每日最新资讯28at.com

生产者只关注消息是否成功的发送到消息处理中间件，消费者只关注能否从消息处理中间件消费到消息。24m28资讯网——每日最新资讯28at.com

消费者可以按组消费，同组内的消费者进行抢占式消费。24m28资讯网——每日最新资讯28at.com

RabbitMq中的优秀实践

1.RabbitMq消息处理机制

24m28资讯网——每日最新资讯28at.com

生产者讲带有指定RoutingKey的消息发送到对应的Exchange上，Exchange通过Binding定义的路由规格，将消息按照BindingKey分发到不同的Queue上，消费者从Queue拉取消息消费。24m28资讯网——每日最新资讯28at.com

• Exchange & RoutingKey & Topic：RoutingKey决定了消息会被发送到哪个Exchange上，这和topic是类似的概念。
• Bind & BindingKey & Group：Exchange根据Binding定义的路由规格，将消息按照BindingKey分发到不同的Queue上，这里可以认为是对应了Group的概念。
• Queue & Group：Queue则是维护了一个Group下的某个队列下的所有消息。

优秀实践

因此如果要以RabbitMq实现基于Topic和Group实现的消息生产和消费的机制，可以将消息定义成以下类似的结构：24m28资讯网——每日最新资讯28at.com

// Exchange: {value="order-update", type="fanout"}// binding1: {value="promotion-service", bindingKey="order.*.paid"}// binding2: {value='inventory-service', bindingKey="order.*"}OrderUpdateMessage(id="msg-0001", orderId = "order-0001"， action="create", ...)OrderUpdateMessage(id="msg-0002", orderId = "order-0001"， action="paid", ...)OrderUpdateMessage(id="msg-0003", orderId = "order-0001"， action="modified", ...)OrderUpdateMessage(id="msg-0004", orderId = "order-0001"， action="delivering", ...)OrderUpdateMessage(id="msg-0005", orderId = "order-0001"， action="delivered", ...)OrderUpdateMessage(id="msg-0006", orderId = "order-0002"， action="paid", ...)

假设此时有promotion-service（1个实例）和inventory-service（2个实例）两个消费者消费消息，则对应的消息消费的结果可能是：24m28资讯网——每日最新资讯28at.com

// inventory-service// Exchange: {value="order-update", type="fanout"}// QueueBinding: {value=Queue('inventory-service'), bindingKey="order.*"}// inventory-service实例1消费到的消息OrderUpdateMessage(id="msg-0001", orderId = "order-0001"， action="create", ...)OrderUpdateMessage(id="msg-0002", orderId = "order-0001"， action="paid", ...)OrderUpdateMessage(id="msg-0006", orderId = "order-0002"， action="paid", ...)// inventory-service实例2消费到的消息OrderUpdateMessage(id="msg-0003", orderId = "order-0001"， action="modified", ...)OrderUpdateMessage(id="msg-0004", orderId = "order-0001"， action="delivering", ...)OrderUpdateMessage(id="msg-0005", orderId = "order-0001"， action="delivered", ...)// promotion-service// Exchange: {value="order-update", type="fanout"}// QueueBinding: {value="promotion-service", bindingKey="order.*.paid"}// promotion-service实例1消费到的消息OrderUpdateMessage(id="msg-0002", orderId = "order-0001"， action="paid", ...)OrderUpdateMessage(id="msg-0006", orderId = "order-0002"， action="paid", ...)

总结

RabbitMQ的Exchange支持不同类型(Direct, Fanout, Topic, Headers)，以及Binding可以对消息以更灵活的通配符的方式将消息分发到对应的Queue上，因此其消息处理机制更加灵活。24m28资讯网——每日最新资讯28at.com

基于Topic的消息发布与消费模式，能够将消费者和生产者完全解耦，相对RabbitMQ中的所支持的灵活处理消息的方式，更加简单且易于理解，这也是Kafka的消息处理机制。24m28资讯网——每日最新资讯28at.com

通过对比不同的中间件的消息处理机制也许能找到更好的实践方式。24m28资讯网——每日最新资讯28at.com

本文链接：http://www.28at.com/showinfo-26-17548-0.html基于Topic的消息发布与消费模式

声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com

上一篇： 数据结构与集合的不解之缘，你了解多少？

下一篇： 解锁多核处理器的力量：探索数据并行化在 Java 8 Stream 中的应用