?

簡(jiǎn)述 ? ?????

?

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

?

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

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

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

?

JMS規(guī)范

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

什么是jms呢

????????jms其實(shí)就是類似于jdbc的一套接口規(guī)范，不同的是他是面向的消息服務(wù)，提供一套標(biāo)準(zhǔn)API接口。大部分廠商都會(huì)參考jms規(guī)范，不過 rocketmq 卻沒有嚴(yán)格遵守jms規(guī)范。

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

基本概念

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

• 點(diǎn)對(duì)點(diǎn)（ Point-to-Point(P2P) ）

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

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

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

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

• 消息隊(duì)列（Queue）

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

• 主題（Topic）

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

• 發(fā)布者（Publisher）

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

• 訂閱者（Subscriber）

????????針對(duì)同一主題的多個(gè)消費(fèi)者

?點(diǎn)對(duì)點(diǎn)

點(diǎn)對(duì)點(diǎn)的關(guān)系圖

發(fā)布訂閱

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

對(duì)象模型

• (1) ConnectionFactory

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

• (2) Destination

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

• (3) Connection

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

• (4) Session

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

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

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

• (6) 消息消費(fèi)者

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

• (7) MessageListener

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

?

消息消費(fèi)

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

○ 同步

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

○ 異步

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

編程實(shí)例

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

舉個(gè)例子：

?

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

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

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

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

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

broker：消息隊(duì)列核心，相當(dāng)于一個(gè)控制中心，負(fù)責(zé)路由消息、保存訂閱和連接、消息確認(rèn)和控制事務(wù)，activemq可以配置多個(gè)

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

public void sendMsg(String queueName){try {//創(chuàng)建一個(gè)消息隊(duì)列（此處也就是在創(chuàng)建Destination）Queue queue = session.createQueue(queueName);//消息生產(chǎn)者M(jìn)essageProducer 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);//提交事務(wù)session.commit();}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}

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

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

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

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

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

發(fā)送消息

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

?

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

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:我是消費(fèi)者，我正在消費(fèi)Msg"+msg.getText()+"--->"+count.getAndIncrement());}else {break;}}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}

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

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

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

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

如果消息不為空告知消息消費(fèi)成功。

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

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

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

?

RocketMQ

簡(jiǎn)介

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

????????大概特點(diǎn)：

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

?

選擇的理由

?rocketmq 的特性

• 強(qiáng)調(diào)集群無單點(diǎn)，可擴(kuò)展，任意一點(diǎn)高可用，水平可擴(kuò)展

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

• 支持上萬個(gè)隊(duì)列，順序消息

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

• 任性定制你的消息過濾

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

• 消息的可靠性（無Buffer，持久化，容錯(cuò)，回溯消費(fèi)）

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

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

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

• 分布式事務(wù)

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

• 消息失敗重試機(jī)制

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

????????針對(duì)consumer的重試，如果消息因?yàn)楦鞣N原因沒有消費(fèi)成功，會(huì)自動(dòng)加入到重試隊(duì)列，一般情況如果是因?yàn)榫W(wǎng)絡(luò)等問題連續(xù)重試也是照樣失敗，所以rocketmq也是采用階梯重試的方式。

• 定時(shí)消費(fèi)

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

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

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

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

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

?

術(shù)語

• NameServer

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

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

• Push Consumer

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

• Pull Consume

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

• Producer Group

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

• Consumer Group

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

• Broker

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

• Message Filter

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

• Master/Slave

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

?

部署方式

物理部署

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

1、接收broker的注冊(cè)，注銷請(qǐng)求

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

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

Broker ：

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

Producer：

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

Consumer:

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

邏輯部署

Producer Group：

用來表示一個(gè)發(fā)送消息應(yīng)用，一個(gè)Producer Group下辦好多個(gè)Producer實(shí)例，可是多臺(tái)機(jī)器，也可以是一臺(tái)機(jī)器的多個(gè)線程，或一個(gè)進(jìn)程的多個(gè)Producer對(duì)象，一個(gè)Producer Group可以發(fā)送多個(gè)Topic消息，Producer Group的作用如下：

1、標(biāo)識(shí)一類Producer（分布式）

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

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

Consumer Group：

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

?

單Master模式

??????只有一個(gè) Master節(jié)點(diǎn)

• 優(yōu)點(diǎn)：配置簡(jiǎn)單，方便部署
• 缺點(diǎn)：這種方式風(fēng)險(xiǎn)較大，一旦Broker重啟或者宕機(jī)時(shí)，會(huì)導(dǎo)致整個(gè)服務(wù)不可用，不建議線上環(huán)境使用

多Master模式

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

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

?

多Master多Slave模式（異步復(fù)制）

????????每個(gè) Master 配置一個(gè) Slave，有多對(duì)Master-Slave， HA，采用異步復(fù)制方式，主備有短暫消息延遲，毫秒級(jí)。

• 優(yōu)點(diǎn)：即使磁盤損壞，消息丟失的非常少，且消息實(shí)時(shí)性不會(huì)受影響，因?yàn)镸aster 宕機(jī)后，消費(fèi)者仍然可以從 Slave消費(fèi)，此過程對(duì)應(yīng)用透明。不需要人工干預(yù)。性能同多 Master 模式幾乎一樣。
• 缺點(diǎn)： Master 宕機(jī)，磁盤損壞情況，會(huì)丟失少量消息。

?

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

????????每個(gè) Master 配置一個(gè) Slave，有多對(duì)Master-Slave， HA采用同步雙寫方式，主備都寫成功，向應(yīng)用返回成功。

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

?

特性使用

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)建一個(gè)Producer的，這里我們看到rocketmq的創(chuàng)建producer很簡(jiǎn)單只輸入一個(gè)Group Name名字就可以。

2、第二步就是設(shè)定Name Server的地址，這里注意兩點(diǎn)，一個(gè)就是nameserver的默認(rèn)端口是9876，另一個(gè)就是多個(gè)nameserver集群用分號(hào)來分割。

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

4、消息創(chuàng)建也很簡(jiǎn)單，第一個(gè)參數(shù)是topic，第二個(gè)就是tags（多個(gè)tag用 || 連接），第三個(gè)參宿是消息內(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");/*** 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開始消費(fèi)還是隊(duì)列尾部開始消費(fèi)<br>* 如果非第一次啟動(dòng)，那么按照上次消費(fèi)的位置繼續(xù)消費(fèi)*/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、可以設(shè)置從哪個(gè)位置開始讀取消息，一般從頭部開始讀取消息，系統(tǒng)中注意去重，即冪等。

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

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

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

?

消息過濾

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

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

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

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

順序消息

一些消息需要按照順序消費(fèi)才有意義。比如： 訂單創(chuàng)建 --> 分批 --> 打包 --> 外發(fā) ... 必須嚴(yán)格按照順序才有意義。rocketmq實(shí)現(xiàn)的方式也很簡(jiǎn)單，只要把這些消息都放到一個(gè)隊(duì)列中就能順序消費(fèi)了。實(shí)際上rocketmq的順序消費(fèi)有兩種方式：一種是普通的順序消費(fèi)（多Master多Slave的異步復(fù)制）；另一種是嚴(yán)格的順序消費(fèi)（多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、首先要保障消息要同時(shí)在一個(gè)topic中

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

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

前面說過 rocketmq 可以同時(shí)支持上萬個(gè)隊(duì)列，這也是為了順序消費(fèi)而考慮的

事務(wù)消息

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

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'

?

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

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

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

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

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

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

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

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

2、同時(shí)張三的賬戶減去10

3、同時(shí)丟出一個(gè)mq消息給rocketmq，兩個(gè)要確保放在一個(gè)db事務(wù)中（此時(shí)的消息只是處于prapared階段，不會(huì)被Consumer所消費(fèi)）

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

5、如果第四部出現(xiàn)了本 Consumer 宕機(jī)，也就是 rocketmq 沒有收到 commit，此刻消息是是未知，所以他會(huì)向任意一臺(tái)Producer 來確認(rèn)當(dāng)前消息的狀態(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ā)送事務(wù)消息例子* */ 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");// 事務(wù)回查最小并發(fā)數(shù)producer.setCheckThreadPoolMinSize(2);// 事務(wù)回查最大并發(fā)數(shù)producer.setCheckThreadPoolMaxSize(2);// 隊(duì)列數(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();} }

本地事務(wù)：

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í)行本地事務(wù)*/ 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)檢查點(diǎn)：

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;/*** 未決事務(wù)，服務(wù)器回查客戶端*/ 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;} }

?

點(diǎn)對(duì)點(diǎn)/廣播

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

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

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

推送/拉取

上面都是消息推送模式，注冊(cè)監(jiān)聽，當(dāng)有消息產(chǎn)生時(shí)就會(huì)實(shí)時(shí)的推送到Consumer進(jìn)行消費(fèi)。

消息拉取方式則相當(dāng)于把主動(dòng)權(quán)交給了應(yīng)用自己，當(dāng)然這樣也給消費(fèi)增加了復(fù)雜性。比如說offset的存儲(chǔ)、定時(shí)拉取等。

阿里給我們提供了一個(gè)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ù)時(shí)間來設(shè)置消費(fèi)進(jìn)度，設(shè)置之前要關(guān)閉這個(gè)訂閱組的所有consumer，設(shè)置完再啟動(dòng)，方可生效。

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

?

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

總結(jié)

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

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