?

簡述 ? ?????

?

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

?

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

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

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

?

JMS規范

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

什么是jms呢

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

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

基本概念

發送者（ Sender）
----?也就是消息的生產者，創建并發送消息的 JMS 客戶端。接收者（ Receiver）?
----?消息消費者，接收訂制消息并按相應業務邏輯進行處理，最終將結果反饋給 mq 的服務端。

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

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

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

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

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

• 消息隊列（Queue）

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

• 主題（Topic）

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

• 發布者（Publisher）

????????同生產者

• 訂閱者（Subscriber）

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

?點對點

點對點的關系圖

發布訂閱

發布訂閱的關系圖

對象模型

• (1) ConnectionFactory

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

• (2) Destination

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

• (3) Connection

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

• (4) Session

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

• (5) 消息的生產者

????????消息生產者由Session創建，并用于將消息發送到Destination。同樣，消息生產者分兩種類型：QueueSender和TopicPublisher。可以調用消息生產者的方法（send或publish方法）發送消息。

• (6) 消息消費者

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

• (7) MessageListener

????????消息監聽器。如果注冊了消息監聽器，一旦消息到達，將自動調用監聽器的onMessage方法。

?

消息消費

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

○ 同步

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

○ 異步

訂閱者或接收者可以注冊為一個消息監聽器。當消息到達之后，系統自動調用監聽器的 onMessage 方法。

編程實例

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

舉個例子：

?

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

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

1. 首先我們需要創建一個連接工廠，當然這里我們需要輸入用戶性和密碼還有就是broker的url
2. 然后我們根據連接工廠創建了一個連接，此刻這個工廠并沒有和broker簡歷連接
3. 調用start方法就和broker建立了連接，這里我大概解釋一下broker
   1. broker：消息隊列核心，相當于一個控制中心，負責路由消息、保存訂閱和連接、消息確認和控制事務，activemq可以配置多個
4. 創建一個session，上面我們提到過所有的消息操作都是與session進行的

   public void sendMsg(String queueName){try {//創建一個消息隊列（此處也就是在創建Destination）Queue queue = session.createQueue(queueName);//消息生產者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();//創建一條消息TextMessage msg = session.createTextMessage(Thread.currentThread().getName()+"productor:生產消息,count:"+num);//發送消息messageProducer.send(msg);//提交事務session.commit();}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}

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

1. 通過session創建一個Destination，我這邊直接就用了queue了
2. 接下來我們需要創建一個消息的生產者
3. 我這邊就循環每1s發送一條消息
4. 這邊看到我們的消息也是用session來創建的，這里面我們用的是文本的消息類型
5. 發送消息
6. 提交這次發送，至此我們的消息就發送到了broker上了，用過activemq的同學都知道，activemq提供了一個很好用的界面可以查到你的消息的狀態，包括是否消費等。

?

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

    public void doMessage(String queueName){try {//創建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();}}

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

1. 通過session創建一個Destination，我這邊直接就用了queue了。
2. 創建了一個Consumer。
3. 做了一個死循環，類似于ServerSocket的accept方法，我們的receive會阻塞到這里，直到有消息。
4. 如果消息不為空告知消息消費成功。

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、注冊了一個監聽接口的實現，當有消息時就調用onMessage的實現，后面就一樣了

?

RocketMQ

簡介

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

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

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

?

選擇的理由

?rocketmq 的特性

• 強調集群無單點，可擴展，任意一點高可用，水平可擴展

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

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

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

• 任性定制你的消息過濾

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

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

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

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

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

• 分布式事務

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

• 消息失敗重試機制

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

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

• 定時消費

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

• 活躍的開源社區

現在rocketmq成為了apache的一款開源產品，活躍度也是不容懷疑的

• 成熟度（經過雙十一考驗）

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

?

術語

• NameServer

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

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

• Push Consumer

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

• Pull Consume

????????Consumer 的一種，應用通常主動調用 Consumer 的拉消息方法從 Broker 拉消息，主動權由應用控制

• Producer Group

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

• Consumer Group

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

• Broker

????????消息中轉的角色，負責存儲消息（實際的存儲是調用的store組件完成的），轉發消息，一般也成為 server，同于?jms 中的provider

• Message Filter

????????可以實現高級的自定義的消息過濾，java編寫

• Master/Slave

????????集群的主從關系，broker 的 name 相同，brokerid=0 的為主，大于 0 的為從

?

部署方式

物理部署

NameServer ：類似云zk的集群，主要是維護了broker的相關內容，進行存取；節點之間無任何數據同步

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

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

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

Broker ：

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

Producer：

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

Consumer:

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

邏輯部署

Producer Group：

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

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

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

3、發送分布式事務消息時，如果Producer中途意外宕機，Broker會主動回調Producer Group內的任意一臺機器來確認事務狀態。

Consumer Group：

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

?

單Master模式

??????只有一個 Master節點

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

多Master模式

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

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

?

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

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

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

?

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

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

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

?

特性使用

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，發送消息* */
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、創建一個Producer的，這里我們看到rocketmq的創建producer很簡單只輸入一個Group Name名字就可以。

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

3、我這邊循環發送了1000個消息。

4、消息創建也很簡單，第一個參數是topic，第二個就是tags（多個tag用 || 連接），第三個參宿是消息內容。

5、調用send方法就能發送成功了（不用像 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>* 如果非第一次啟動，那么按照上次消費的位置繼續消費*/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、可以設置從哪個位置開始讀取消息，一般從頭部開始讀取消息，系統中注意去重，即冪等。

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

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

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

?

消息過濾

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

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

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

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

順序消息

一些消息需要按照順序消費才有意義。比如： 訂單創建 --> 分批 --> 打包 --> 外發 ... 必須嚴格按照順序才有意義。rocketmq實現的方式也很簡單，只要把這些消息都放到一個隊列中就能順序消費了。實際上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，發送順序消息*/
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、要保障要發送的消息有相同的tag

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

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

事務消息

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

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

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

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、鎖定張三的賬戶，同時通過網絡鎖定李四的賬戶（可以理解成凍結金額）

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

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

4、通過網絡向李四的賬戶增加10塊

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

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

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

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

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

4、如果本地事務執行成功則向 rocketmq 發送 commit

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

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;/*** 發送事務消息例子* */
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");// 事務回查最小并發數producer.setCheckThreadPoolMinSize(2);// 事務回查最大并發數producer.setCheckThreadPoolMaxSize(2);// 隊列數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;/*** 執行本地事務*/
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;}
}

回調檢查點：

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;}
}

?

點對點/廣播

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

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

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

推送/拉取

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

消息拉取方式則相當于把主動權交給了應用自己，當然這樣也給消費增加了復雜性。比如說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;}}

消息回溯

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

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

?

轉自：https://my.oschina.net/izhangll/blog/1581254，有作部分調整。