閑話

朋友們，好久不見，不知道你們最近怎樣，但相信你們一定都挺好。已經有一段時間沒有更新了，個中原因不好細說，但是歸根結底也許是自己懶。這個不好，大家不要學。今天主要就是想分享一下關于消息處理機制的一些想法。BLc28資訊網——每日最新資訊28at.com

基本概念

BLc28資訊網——每日最新資訊28at.com

1.Topic

同一個topic下消息的格式一致，例如topic為order-update-message消息的格式都是一個統一的OrderUpdateMessage的結構BLc28資訊網——每日最新資訊28at.com

2.key主鍵

同一主鍵下的消息列表具有順序性，例如key為訂單號order-0001的消息列表(Queue)下，可能包含的消息列表(Queue)如下：BLc28資訊網——每日最新資訊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，消息消費的結果可能如下：BLc28資訊網——每日最新資訊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的消息處理機制就是以這樣的形式實現的。BLc28資訊網——每日最新資訊28at.com

4.優勢 

生產者和消費者完全解耦，生產者無需關注是否有消費者在消費，消費者也無需知道生產者是否在生成新的消息。BLc28資訊網——每日最新資訊28at.com

生產者只關注消息是否成功的發送到消息處理中間件，消費者只關注能否從消息處理中間件消費到消息。BLc28資訊網——每日最新資訊28at.com

消費者可以按組消費，同組內的消費者進行搶占式消費。BLc28資訊網——每日最新資訊28at.com

RabbitMq中的優秀實踐

1.RabbitMq消息處理機制

BLc28資訊網——每日最新資訊28at.com

生產者講帶有指定RoutingKey的消息發送到對應的Exchange上，Exchange通過Binding定義的路由規格，將消息按照BindingKey分發到不同的Queue上，消費者從Queue拉取消息消費。BLc28資訊網——每日最新資訊28at.com

• Exchange & RoutingKey & Topic：RoutingKey決定了消息會被發送到哪個Exchange上，這和topic是類似的概念。
• Bind & BindingKey & Group：Exchange根據Binding定義的路由規格，將消息按照BindingKey分發到不同的Queue上，這里可以認為是對應了Group的概念。
• Queue & Group：Queue則是維護了一個Group下的某個隊列下的所有消息。

優秀實踐

因此如果要以RabbitMq實現基于Topic和Group實現的消息生產和消費的機制，可以將消息定義成以下類似的結構：BLc28資訊網——每日最新資訊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個實例）兩個消費者消費消息，則對應的消息消費的結果可能是：BLc28資訊網——每日最新資訊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上，因此其消息處理機制更加靈活。BLc28資訊網——每日最新資訊28at.com

基于Topic的消息發布與消費模式，能夠將消費者和生產者完全解耦，相對RabbitMQ中的所支持的靈活處理消息的方式，更加簡單且易于理解，這也是Kafka的消息處理機制。BLc28資訊網——每日最新資訊28at.com

通過對比不同的中間件的消息處理機制也許能找到更好的實踐方式。BLc28資訊網——每日最新資訊28at.com

本文鏈接：http://www.www897cc.com/showinfo-26-17548-0.html基于Topic的消息發布與消費模式

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇： 數據結構與集合的不解之緣，你了解多少？

下一篇： 解鎖多核處理器的力量：探索數據并行化在 Java 8 Stream 中的應用