?

簡述 ? ?????

?

前些天發(fā)現(xiàn)了一個巨牛的人工智能學習網(wǎng)站，通俗易懂，風趣幽默，忍不住分享一下給大家。點擊跳轉(zhuǎn)到教程。

?

????????分布式消息中間件，主要是實現(xiàn)分布式系統(tǒng)中解耦、異步消息、流量銷鋒、日志處理等場景。生產(chǎn)中用的最多的消息隊列有Activemq，rabbitmq，kafka，rocketmq等。

? ? ? ? 以 Jms 規(guī)范和 rocketmq 為主來分享。版本基于 3.2.6 。

????????主要分享：JMS規(guī)范、Rocketmq的介紹、部署方式、特性的一些使用。

?

JMS規(guī)范

? ? ? ? rocketmq雖然不完全基于jms規(guī)范，但參考了jms規(guī)范和 CORBA Notification 規(guī)范，且青出于藍而勝于藍。

什么是jms呢

????????jms其實就是類似于jdbc的一套接口規(guī)范，不同的是他是面向的消息服務，提供一套標準API接口。大部分廠商都會參考jms規(guī)范，不過 rocketmq 卻沒有嚴格遵守jms規(guī)范。

???????常見的jms廠商有：IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，還有APACHE開源的ActiveMQ。京東商城采用的就是 Activemq 。

基本概念

發(fā)送者（ Sender） ----?也就是消息的生產(chǎn)者，創(chuàng)建并發(fā)送消息的 JMS 客戶端。接收者（ Receiver）? ----?消息消費者，接收訂制消息并按相應業(yè)務邏輯進行處理，最終將結(jié)果反饋給 mq 的服務端。

• 點對點（ Point-to-Point(P2P) ）

????????點對點是一對一的關(guān)系，一個消息發(fā)出只有一個接受者所處理。每個消息都被發(fā)送到一個特定的隊列，接收者從隊列中獲取消息。隊列保留著消息，直到他們被消費或超時。

• 發(fā)布訂閱（ Publish/Subscribe(Pub/Sub) ）

????????1、客戶端將消息發(fā)送到主題。多個發(fā)布者將消息發(fā)送到Topic,系統(tǒng)將這些消息傳遞給多個訂閱者。

????????2、如果你希望發(fā)送的消息不被做任何處理、或者被一個消息者處理、或者可以被多個消費者處理的話，那么可以采用Pub/Sub模型

• 消息隊列（Queue）

????????一個容納那些被發(fā)送的等待閱讀的消息的區(qū)域。與隊列名字所暗示的意思不同，消息的接受順序并不一定要與消息的發(fā)送順序相同。一旦一個消息被閱讀，該消息將被從隊列中移走。

• 主題（Topic）

????????一種支持發(fā)送消息給多個訂閱者的機制。

• 發(fā)布者（Publisher）

????????同生產(chǎn)者

• 訂閱者（Subscriber）

????????針對同一主題的多個消費者

?點對點

點對點的關(guān)系圖

發(fā)布訂閱

發(fā)布訂閱的關(guān)系圖

對象模型

• (1) ConnectionFactory

????????創(chuàng)建Connection對象的工廠，針對兩種不同的jms消息模型，分別有QueueConnectionFactory和TopicConnectionFactory兩種（基于點對點和和發(fā)布訂閱的兩種方式分別創(chuàng)建連接工廠的）。可以通過JNDI來查找ConnectionFactory對象。

• (2) Destination

????????Destination 是消息生產(chǎn)者的消息發(fā)送目標，或者是消息消費者的消息來源。對于消息生產(chǎn)者來說，它的Destination是某個隊列（Queue）或某個主題（Topic）;對于消息消費者來說，它的Destination也是某個隊列或主題（即消息來源）。所以，Destination實際上就是兩種類型的對象：Queue、Topic可以通過JNDI來查找Destination。

• (3) Connection

????????Connection表示在客戶端和JMS系統(tǒng)之間建立的鏈接（對TCP/IP socket的包裝）。Connection可以產(chǎn)生一個或多個Session。跟ConnectionFactory一樣，Connection也有兩種類型：QueueConnection和TopicConnection。

• (4) Session

????????Session是我們操作消息的接口。可以通過session創(chuàng)建生產(chǎn)者、消費者、消息等。Session提供了事務的功能。當我們需要使用session發(fā)送/接收多個消息時，可以將這些發(fā)送/接收動作放到一個事務中。同樣，也分QueueSession和TopicSession。

• (5) 消息的生產(chǎn)者

????????消息生產(chǎn)者由Session創(chuàng)建，并用于將消息發(fā)送到Destination。同樣，消息生產(chǎn)者分兩種類型：QueueSender和TopicPublisher。可以調(diào)用消息生產(chǎn)者的方法（send或publish方法）發(fā)送消息。

• (6) 消息消費者

????????消息消費者由Session創(chuàng)建，用于接收被發(fā)送到Destination的消息。兩種類型：QueueReceiver和TopicSubscriber。可分別通過session的createReceiver(Queue)或createSubscriber(Topic)來創(chuàng)建。當然，也可以session的creatDurableSubscriber方法來創(chuàng)建持久化的訂閱者。

• (7) MessageListener

????????消息監(jiān)聽器。如果注冊了消息監(jiān)聽器，一旦消息到達，將自動調(diào)用監(jiān)聽器的onMessage方法。

?

消息消費

在JMS中，消息的產(chǎn)生和消息是異步的。對于消費來說，JMS的消息者可以通過兩種方式來消費消息。

○ 同步

訂閱者或接收者調(diào)用receive方法來接收消息，receive方法在能夠接收到消息之前（或超時之前）將一直阻塞

○ 異步

訂閱者或接收者可以注冊為一個消息監(jiān)聽器。當消息到達之后，系統(tǒng)自動調(diào)用監(jiān)聽器的 onMessage 方法。

編程實例

通過 activemq 的部分代碼來簡單說明一下上面說到的一些JMS規(guī)范

舉個例子：

?

public void init(){try {//創(chuàng)建一個鏈接工廠（用戶名，密碼，broker的url地址）connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);//從工廠中創(chuàng)建一個鏈接connection = connectionFactory.createConnection();//開啟鏈接connection.start();//創(chuàng)建一個會話session = connection.createSession(true,Session.SESSION_TRANSACTED);} catch (JMSException e) {e.printStackTrace();}}

????公共部分：也就是說不管你是消息的生產(chǎn)者還是消息的消費者都需要這些步驟

首先我們需要創(chuàng)建一個連接工廠，當然這里我們需要輸入用戶性和密碼還有就是broker的url

然后我們根據(jù)連接工廠創(chuàng)建了一個連接，此刻這個工廠并沒有和broker簡歷連接

調(diào)用start方法就和broker建立了連接，這里我大概解釋一下broker

broker：消息隊列核心，相當于一個控制中心，負責路由消息、保存訂閱和連接、消息確認和控制事務，activemq可以配置多個

創(chuàng)建一個session，上面我們提到過所有的消息操作都是與session進行的

public void sendMsg(String queueName){try {//創(chuàng)建一個消息隊列（此處也就是在創(chuàng)建Destination）Queue queue = session.createQueue(queueName);//消息生產(chǎn)者MessageProducer messageProducer = null;if(threadLocal.get()!=null){messageProducer = threadLocal.get();}else{messageProducer = session.createProducer(queue);threadLocal.set(messageProducer);}while(true){Thread.sleep(1000);int num = count.getAndIncrement();//創(chuàng)建一條消息TextMessage msg = session.createTextMessage(Thread.currentThread().getName()+"productor:生產(chǎn)消息,count:"+num);//發(fā)送消息messageProducer.send(msg);//提交事務session.commit();}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}

生產(chǎn)：配置完上面的公共部分我們就迫不及待的把消息生產(chǎn)出來吧，我這邊說的是點對點的方式

通過session創(chuàng)建一個Destination，我這邊直接就用了queue了

接下來我們需要創(chuàng)建一個消息的生產(chǎn)者

我這邊就循環(huán)每1s發(fā)送一條消息

這邊看到我們的消息也是用session來創(chuàng)建的，這里面我們用的是文本的消息類型

發(fā)送消息

提交這次發(fā)送，至此我們的消息就發(fā)送到了broker上了，用過activemq的同學都知道，activemq提供了一個很好用的界面可以查到你的消息的狀態(tài)，包括是否消費等。

?

消費：消費我們上面也提到了兩種方式，同步和異步，我這邊準備了兩份代碼分別說明了一下

public void doMessage(String queueName){try {//創(chuàng)建DestinationQueue queue = session.createQueue(queueName);MessageConsumer consumer = null;while(true){Thread.sleep(1000);TextMessage msg = (TextMessage) consumer.receive();if(msg!=null) {msg.acknowledge();System.out.println(Thread.currentThread().getName()+": Consumer:我是消費者，我正在消費Msg"+msg.getText()+"--->"+count.getAndIncrement());}else {break;}}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}

同步：可以看到消息會一直阻塞到有消息才會繼續(xù)

通過session創(chuàng)建一個Destination，我這邊直接就用了queue了。

創(chuàng)建了一個Consumer。

做了一個死循環(huán)，類似于ServerSocket的accept方法，我們的receive會阻塞到這里，直到有消息。

如果消息不為空告知消息消費成功。

consumer.setMessageListener(MessageListener { public void onMessage(Message msg) { try { String message = ((TextMessage) msg).getText(); if(msg != null){msg.acknowledgeSystem.out.println("成功消費消息："+message);} } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } );

異步：前兩部和上面是一樣的，我們從第三步說起

3、注冊了一個監(jiān)聽接口的實現(xiàn)，當有消息時就調(diào)用onMessage的實現(xiàn)，后面就一樣了

?

RocketMQ

簡介

????????rocketmq是阿里巴巴開源的一款分布式的消息中間件，源于jms規(guī)范，但是不遵守jms規(guī)范。rocketmq天生就是分布式的，可以說是broker、provider、consumer等各種分布式。

????????大概特點：

• 能夠保證嚴格的消息順序（需要集群的支持）
• 提供豐富的消息拉取模式（可以任意定義你的拉取方式，exmaple中也提供了一個很好的例子）
• 高效的訂閱者水平擴展能力（通過一個consumerGroup的方式做到consumer的方便擴容）
• 實時的消息訂閱機制（消息的實時推送，類似于上面咱們的異步消費的方式）
• 億級消息堆積能力（輕松完成系統(tǒng)銷鋒）

?

選擇的理由

?rocketmq 的特性

• 強調(diào)集群無單點，可擴展，任意一點高可用，水平可擴展

????????方便集群配置，而且容易擴展（橫向和縱向），通過slave的方式每一點都可以實現(xiàn)高可用

• 支持上萬個隊列，順序消息

????????順序消費是實現(xiàn)在同一隊列的，如果高并發(fā)的情況就需要隊列的支持，rocketmq可以滿足上萬個隊列同事存在

• 任性定制你的消息過濾

????????rocketmq提供了兩種類型的消息過濾，也可以說三種可以通過topic進行消息過濾、可以通過tag進行消息過濾、還可以通過filter的方式任意定制過濾

• 消息的可靠性（無Buffer，持久化，容錯，回溯消費）

????????消息無buffer就不用擔心buffer回滿的情況，rocketmq的所有消息都是持久化的，生產(chǎn)者本身可以進行錯誤重試，發(fā)送者也會按照時間階梯的方式進行消息重發(fā)，消息回溯說的是可以按照指定的時間進行消息的重新消費，既可以向前也可以向后（前提條件是要注意消息的擦除時間）

• 海量消息堆積能力，消息堆積后，寫入低延遲

????????針對于provider需要配合部署方式，對于consumer，如果是集群方式一旦master返現(xiàn)消息堆積會向consumer下發(fā)一個重定向指令，此時consumer就可以從slave進行數(shù)據(jù)消費了

• 分布式事務

????????我個人感覺 rocketmq3.2.6 對這一塊說的不是很清晰，而且官方也說現(xiàn)在這塊存在缺陷（會令系統(tǒng)pagecache過多），所以線上建議還是少用為好，這塊后面有列子。

• 消息失敗重試機制

????????針對provider的重試，當消息發(fā)送到選定的broker時如果出現(xiàn)失敗會自動選擇其他的broker進行重發(fā)，默認重試三次，當然重試次數(shù)要在消息發(fā)送的超時時間范圍內(nèi)。

????????針對consumer的重試，如果消息因為各種原因沒有消費成功，會自動加入到重試隊列，一般情況如果是因為網(wǎng)絡等問題連續(xù)重試也是照樣失敗，所以rocketmq也是采用階梯重試的方式。

• 定時消費

出了上面的配置，在發(fā)送消息是也可以針對message設置setDelayTimeLevel

• 活躍的開源社區(qū)

現(xiàn)在rocketmq成為了apache的一款開源產(chǎn)品，活躍度也是不容懷疑的

• 成熟度（經(jīng)過雙十一考驗）

針對本身的成熟度，我們看看這么多年的雙十一就可想而知了

?

術(shù)語

• NameServer

????????可以理解成類似于zk的一個注冊中心，而且rocketmq最初也是基于zk作為注冊中心的，現(xiàn)在相當于為rocketmq自定義了一個注冊中心，代碼不超過1000行。RocketMQ 有多種配置方式可以令客戶端找到 Name Server, 然后通過 Name Server 再找到 Broker，分別如下，

優(yōu)先級由高到低，高優(yōu)先級會覆蓋低優(yōu)先級。客戶端提供 http 和 ip + 端口號的兩種方式，推薦使用 http 的方式可以實現(xiàn)nameserver 的熱部署

• Push Consumer

????????Consumer 的一種，應用通常通過 Consumer 對象注冊一個 Listener 接口，一旦收到消息，Consumer 對象立刻回調(diào) Listener 接口方法，類似于 activemq 的方式

• Pull Consume

????????Consumer 的一種，應用通常主動調(diào)用 Consumer 的拉消息方法從 Broker 拉消息，主動權(quán)由應用控制

• Producer Group

????????一類producer的集合名稱，這類 producer 通常發(fā)送一類消息，且發(fā)送邏輯一致

• Consumer Group

????????同上，consumer的集合名稱

• Broker

????????消息中轉(zhuǎn)的角色，負責存儲消息（實際的存儲是調(diào)用的store組件完成的），轉(zhuǎn)發(fā)消息，一般也成為 server，同于?jms 中的provider

• Message Filter

????????可以實現(xiàn)高級的自定義的消息過濾，java編寫

• Master/Slave

????????集群的主從關(guān)系，broker 的 name 相同，brokerid=0 的為主，大于 0 的為從

?

部署方式

物理部署

NameServer ：類似云zk的集群，主要是維護了broker的相關(guān)內(nèi)容，進行存取；節(jié)點之間無任何數(shù)據(jù)同步

1、接收broker的注冊，注銷請求

2、Producer獲取topic下所有的BrokerQueue，put消息

3、Consumer獲取topic下所有的BrokerQueue，get消息

Broker ：

部署相對復雜，Broker分為Master與Slave，一個Master可以對應多個Slave，但是一個Slave只能對應Master。Master和Slave的對應關(guān)系通過制定相同的BrokerName來確定，通過制定BrokerId來區(qū)分主從，如果是0則為Master，如果大于0則為Slave。Master也可以部署多個。每個Broker與Name Server集群中的所有節(jié)點建立長連接，定時注冊Topic信息到所有的NameServer

Producer：

與Name sever集群中的其中一個節(jié)點（隨意選擇）建立長連接，定期的從Name Server取Topic路由信息，并向提供Topic服務的Master 建立長連接，且定時向Master發(fā)送心跳。Producer完全無狀態(tài)，可以集群部署。

Consumer:

與Name Server集群中的其中一個節(jié)點(隨機選擇)建立長連接,定期從Name Server取Topic路由信息，并向提供Topic的Master、Slave簡歷長連接，且定時向Master、Slave發(fā)送心跳，Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，訂閱規(guī)則有Broker配置決定。

邏輯部署

Producer Group：

用來表示一個發(fā)送消息應用，一個Producer Group下辦好多個Producer實例，可是多臺機器，也可以是一臺機器的多個線程，或一個進程的多個Producer對象，一個Producer Group可以發(fā)送多個Topic消息，Producer Group的作用如下：

1、標識一類Producer（分布式）

2、可以通過運維工具查詢這個發(fā)送消息應用有多少個Producer

3、發(fā)送分布式事務消息時，如果Producer中途意外宕機，Broker會主動回調(diào)Producer Group內(nèi)的任意一臺機器來確認事務狀態(tài)。

Consumer Group：

表示一個消費消息應用，一個Consumer Group下包含多個Consumer實例，可以是多臺機器，也可是多個進程，或者是一個進程的多個Consumer對象。一個Consumer Group下的多個Consumer以均攤方式消費消息。如果設置為廣播方式，那么這個Consumer Group下的每個實例都消費全量數(shù)據(jù)。

?

單Master模式

??????只有一個 Master節(jié)點

• 優(yōu)點：配置簡單，方便部署
• 缺點：這種方式風險較大，一旦Broker重啟或者宕機時，會導致整個服務不可用，不建議線上環(huán)境使用

多Master模式

??????一個集群無 Slave，全是 Master，例如 2 個 Master 或者 3 個 Master

• 優(yōu)點：配置簡單，單個Master 宕機或重啟維護對應用無影響，在磁盤配置為RAID10 時，即使機器宕機不可恢復情況下，由與 RAID10磁盤非常可靠，消息也不會丟（異步刷盤丟失少量消息，同步刷盤一條不丟）。性能最高。多 Master 多 Slave 模式，異步復制
• 缺點：單臺機器宕機期間，這臺機器上未被消費的消息在機器恢復之前不可訂閱，消息實時性會受到受到影響

?

多Master多Slave模式（異步復制）

????????每個 Master 配置一個 Slave，有多對Master-Slave， HA，采用異步復制方式，主備有短暫消息延遲，毫秒級。

• 優(yōu)點：即使磁盤損壞，消息丟失的非常少，且消息實時性不會受影響，因為Master 宕機后，消費者仍然可以從 Slave消費，此過程對應用透明。不需要人工干預。性能同多 Master 模式幾乎一樣。
• 缺點： Master 宕機，磁盤損壞情況，會丟失少量消息。

?

多Master多Slave模式（同步雙寫）

????????每個 Master 配置一個 Slave，有多對Master-Slave， HA采用同步雙寫方式，主備都寫成功，向應用返回成功。

• 優(yōu)點：數(shù)據(jù)與服務都無單點， Master宕機情況下，消息無延遲，服務可用性與數(shù)據(jù)可用性都非常高
• 缺點：性能比異步復制模式略低，大約低 10%左右，發(fā)送單個消息的 RT會略高。目前主宕機后，備機不能自動切換為主機，后續(xù)會支持自動切換功能

?

特性使用

Quick start

Producer：

mport com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.client.producer.DefaultMQProducer; import com.alibaba.rocketmq.client.producer.SendResult; import com.alibaba.rocketmq.common.message.Message;/*** Producer，發(fā)送消息* */ public class Producer {public static void main(String[] args) throws MQClientException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer("pay_topic_01");producer.setNamesrvAddr("100.8.8.88:9876");producer.start();for (int i = 0; i < 1000; i++) {try {Message msg = new Message("TopicTest",// topic"TagA",// tag("Hello RocketMQ " + i).getBytes()// body);SendResult sendResult = producer.send(msg);System.out.println(sendResult);}catch (Exception e) {e.printStackTrace();Thread.sleep(1000);}}producer.shutdown();} }

1、創(chuàng)建一個Producer的，這里我們看到rocketmq的創(chuàng)建producer很簡單只輸入一個Group Name名字就可以。

2、第二步就是設定Name Server的地址，這里注意兩點，一個就是nameserver的默認端口是9876，另一個就是多個nameserver集群用分號來分割。

3、我這邊循環(huán)發(fā)送了1000個消息。

4、消息創(chuàng)建也很簡單，第一個參數(shù)是topic，第二個就是tags（多個tag用 || 連接），第三個參宿是消息內(nèi)容。

5、調(diào)用send方法就能發(fā)送成功了（不用像 actimemq， 還需要commit）。

?

Consumer：

import java.util.List;import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently; import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere; import com.alibaba.rocketmq.common.message.MessageExt;/*** Consumer，訂閱消息*/ public class Consumer {public static void main(String[] args) throws InterruptedException, MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");consumer.setNamesrvAddr("100.8.8.88:9876");/*** 設置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費<br>* 如果非第一次啟動，那么按照上次消費的位置繼續(xù)消費*/consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("TopicTest", "*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println("Consumer Started.");} }

1、前兩步和Producer是一樣的

2、可以設置從哪個位置開始讀取消息，一般從頭部開始讀取消息，系統(tǒng)中注意去重，即冪等。

3、訂閱topic，第一個參數(shù)是topic名字，第二個是tag，如果為 * 的就是全部消息

4、注冊一個監(jiān)聽，如果有消息就會實時的推送到Consumer，調(diào)用consumeMessage進行消費，這里我們看到msgs是一個List，默認每次推送的是一條消息。

5、進行消息的消費邏輯，消費成功后會返回 CONSUME_SUCCESS 狀態(tài)

?

消息過濾

RocketMq的消息過濾是從訂閱的時候開始的，我們看到剛才的例子都是通過topic的tags進行的過濾，這個要求Producer發(fā)送的時候指定tags，這個和前面有點矛盾，但是前面只是進行了分組，并未進行過濾。Consumer在訂閱消費的時候指定了tags才能對消息進行過濾，這種是簡單的過濾方式，不過也可以滿足我們大部分的消息過濾。更高級的過濾如下：

1、前面和后面部分不變，紅色框部分需要指定一個過濾類，之前這里是 tags

2、我們看到所有的過濾類都要直接或者間接實現(xiàn)MessageFilter接口，并且需要覆蓋match方法

3、在方法里面就可以寫自己的過濾邏輯了，這個地方出了用事先制定的屬性也可以反序列化這些消息內(nèi)容進行消息解析，針對消息體的過濾

順序消息

一些消息需要按照順序消費才有意義。比如： 訂單創(chuàng)建 --> 分批 --> 打包 --> 外發(fā) ... 必須嚴格按照順序才有意義。rocketmq實現(xiàn)的方式也很簡單，只要把這些消息都放到一個隊列中就能順序消費了。實際上rocketmq的順序消費有兩種方式：一種是普通的順序消費（多Master多Slave的異步復制）；另一種是嚴格的順序消費（多Master多Slave的同步雙寫）。

import java.util.List;import com.alibaba.rocketmq.client.exception.MQBrokerException; import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.client.producer.DefaultMQProducer; import com.alibaba.rocketmq.client.producer.MQProducer; import com.alibaba.rocketmq.client.producer.MessageQueueSelector; import com.alibaba.rocketmq.client.producer.SendResult; import com.alibaba.rocketmq.common.message.Message; import com.alibaba.rocketmq.common.message.MessageQueue; import com.alibaba.rocketmq.remoting.exception.RemotingException;/*** Producer，發(fā)送順序消息*/ public class Producer {public static void main(String[] args) {try {MQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");producer.setNamesrvAddr("100.8.8.88:9876");producer.start();String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE" };for (int i = 0; i < 100; i++) {// 訂單ID相同的消息要有序int orderId = i % 10;Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,("Hello RocketMQ " + i).getBytes());SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Integer id = (Integer) arg;int index = id % mqs.size();return mqs.get(index);}}, orderId);System.out.println(sendResult);}producer.shutdown();}catch (MQClientException e) {e.printStackTrace();}catch (RemotingException e) {e.printStackTrace();}catch (MQBrokerException e) {e.printStackTrace();}catch (InterruptedException e) {e.printStackTrace();}} }

1、首先要保障消息要同時在一個topic中

2、要保障要發(fā)送的消息有相同的tag

3、在發(fā)送時要保障將數(shù)據(jù)發(fā)送到同一個隊列（queue），我們這里采用的取模的方式

前面說過 rocketmq 可以同時支持上萬個隊列，這也是為了順序消費而考慮的

事務消息

比如有兩個賬戶：張三、李四，張三要給李四轉(zhuǎn)10塊錢。以下都在同一個事務中進行，鎖定是通過事務來完成的

1、鎖定張三和李四的賬戶

2、判斷張三的賬戶是否大于等于10塊錢，如果大于等于則繼續(xù)，小于則返回。（只討論大于等于的）

3、從張三的賬戶上減去10塊

4、向李四的賬戶增加10塊

5、解鎖賬戶完成交易

update account set amount = amount - 100 where userNo='zhangsan' and amount >=10

update account set amount = amount + 100 where userNo='lisi'

?

分布式事務就要考慮到兩個用戶賬戶的一致性，從分布式的角度來分析一下

1、鎖定張三的賬戶，同時通過網(wǎng)絡鎖定李四的賬戶（可以理解成凍結(jié)金額）

2、判斷張三的賬戶是否大于等于10塊錢，如果大于等于則繼續(xù)，小于則返回（只討論大于等于的）

3、從張三的賬戶上減去10塊

4、通過網(wǎng)絡向李四的賬戶增加10塊

5、解鎖張三賬戶完成交易，通過網(wǎng)絡解鎖李四的賬戶，時間基本上是累計的

通過rocketmq怎么實現(xiàn)呢，首先要分清角色，張三為事務的發(fā)起者 = 消息的發(fā)送者，李四就是消息的消費者了。rocketmq可以理解成中間賬戶，默認 Consumer 都會成功，如果不成功官方推薦人工介入。

1、判斷張三的賬戶金額大于10

2、同時張三的賬戶減去10

3、同時丟出一個mq消息給rocketmq，兩個要確保放在一個db事務中（此時的消息只是處于prapared階段，不會被Consumer所消費）

4、如果本地事務執(zhí)行成功則向 rocketmq 發(fā)送 commit

5、如果第四部出現(xiàn)了本 Consumer 宕機，也就是 rocketmq 沒有收到 commit，此刻消息是是未知，所以他會向任意一臺Producer 來確認當前消息的狀態(tài)

6、從此保障了本地賬戶和 rocketmq 的一致性

中控如下：

import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.client.producer.SendResult; import com.alibaba.rocketmq.client.producer.TransactionCheckListener; import com.alibaba.rocketmq.client.producer.TransactionMQProducer; import com.alibaba.rocketmq.common.message.Message;/*** 發(fā)送事務消息例子* */ public class TransactionProducer {public static void main(String[] args) throws MQClientException, InterruptedException {TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");producer.setNamesrvAddr("100.8.8.88:9876");// 事務回查最小并發(fā)數(shù)producer.setCheckThreadPoolMinSize(2);// 事務回查最大并發(fā)數(shù)producer.setCheckThreadPoolMaxSize(2);// 隊列數(shù)producer.setCheckRequestHoldMax(2000);producer.setTransactionCheckListener(transactionCheckListener);producer.start();String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE" };TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();for (int i = 0; i < 100; i++) {try {Message msg =new Message("TopicTest", tags[i % tags.length], "KEY" + i,("Hello RocketMQ " + i).getBytes());SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);System.out.println(sendResult);}catch (MQClientException e) {e.printStackTrace();}}for (int i = 0; i < 100000; i++) {Thread.sleep(1000);}producer.shutdown();} }

本地事務：

import java.util.concurrent.atomic.AtomicInteger;import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter; import com.alibaba.rocketmq.client.producer.LocalTransactionState; import com.alibaba.rocketmq.common.message.Message;/*** 執(zhí)行本地事務*/ public class TransactionExecuterImpl implements LocalTransactionExecuter {private AtomicInteger transactionIndex = new AtomicInteger(1);@Overridepublic LocalTransactionState executeLocalTransactionBranch(final Message msg, final Object arg) {int value = transactionIndex.getAndIncrement();if (value == 0) {throw new RuntimeException("Could not find db");}else if ((value % 5) == 0) {return LocalTransactionState.ROLLBACK_MESSAGE;}else if ((value % 4) == 0) {return LocalTransactionState.COMMIT_MESSAGE;}return LocalTransactionState.UNKNOW;} }

回調(diào)檢查點：

import java.util.concurrent.atomic.AtomicInteger;import com.alibaba.rocketmq.client.producer.LocalTransactionState; import com.alibaba.rocketmq.client.producer.TransactionCheckListener; import com.alibaba.rocketmq.common.message.MessageExt;/*** 未決事務，服務器回查客戶端*/ public class TransactionCheckListenerImpl implements TransactionCheckListener {private AtomicInteger transactionIndex = new AtomicInteger(0);@Overridepublic LocalTransactionState checkLocalTransactionState(MessageExt msg) {System.out.println("server checking TrMsg " + msg.toString());int value = transactionIndex.getAndIncrement();if ((value % 6) == 0) {throw new RuntimeException("Could not find db");}else if ((value % 5) == 0) {return LocalTransactionState.ROLLBACK_MESSAGE;}else if ((value % 4) == 0) {return LocalTransactionState.COMMIT_MESSAGE;}return LocalTransactionState.UNKNOW;} }

?

點對點/廣播

點對點、發(fā)布訂閱兩種模式，在 consumer 里面配置 MessageModel 即可。

需要注意的是：如果配置了發(fā)布訂閱模式，那么 Consumer 的負載均衡將不生效（Consumer Name）

//發(fā)布訂閱consumer.setMessageModel(MessageModel.BROADCASTING);//集群消費（默認）//consumer.setMessageModel(MessageModel.CLUSTERING);

推送/拉取

上面都是消息推送模式，注冊監(jiān)聽，當有消息產(chǎn)生時就會實時的推送到Consumer進行消費。

消息拉取方式則相當于把主動權(quán)交給了應用自己，當然這樣也給消費增加了復雜性。比如說offset的存儲、定時拉取等。

阿里給我們提供了一個demo（文件夾名是simple），可以參考下。

import java.util.HashMap; import java.util.Map; import java.util.Set;import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; import com.alibaba.rocketmq.client.consumer.PullResult; import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.common.message.MessageQueue;/*** PullConsumer，訂閱消息*/ public class PullConsumer {private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();public static void main(String[] args) throws MQClientException {DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");consumer.start();Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");for (MessageQueue mq : mqs) {System.out.println("Consume from the queue: " + mq);SINGLE_MQ: while (true) {try {PullResult pullResult =consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);System.out.println(pullResult);putMessageQueueOffset(mq, pullResult.getNextBeginOffset());switch (pullResult.getPullStatus()) {case FOUND:// TODObreak;case NO_MATCHED_MSG:break;case NO_NEW_MSG:break SINGLE_MQ;case OFFSET_ILLEGAL:break;default:break;}}catch (Exception e) {e.printStackTrace();}}}consumer.shutdown();}private static void putMessageQueueOffset(MessageQueue mq, long offset) {offseTable.put(mq, offset);}private static long getMessageQueueOffset(MessageQueue mq) {Long offset = offseTable.get(mq);if (offset != null)return offset;return 0;}}

消息回溯

根據(jù)時間來設置消費進度，設置之前要關(guān)閉這個訂閱組的所有consumer，設置完再啟動，方可生效。

• 回溯消費是指 Consumer 已經(jīng)消費成功的消息，由于業(yè)務上需求需要重新消費，Broker 在Consumer 投遞成功消息后，消息仍然需要保留。并且重新消費一般是按照時間維度，例如由于 Consumer 系統(tǒng)故障，恢復后需要重新消費 1 小時前的數(shù)據(jù)，?Broker 要提供一種機制，可以按照時間維度來回退消費
• RocketMQ 支持按照時間回溯消費，時間維度精確到毫秒，可以向前回溯，也可以向后回溯
• 操作： mqadmin resetOffsetByTime

?

轉(zhuǎn)自：https://my.oschina.net/izhangll/blog/1581254，有作部分調(diào)整。

總結(jié)

以上是生活随笔為你收集整理的分布式消息中间件 ： Rocketmq的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。