一、簡(jiǎn)要介紹

• 開(kāi)源AMQP實(shí)現(xiàn)，Erlang語(yǔ)言編寫，支持多種客戶端

• 分布式、高可用、持久化、可靠、安全

• 支持多種協(xié)議：AMQP、STOMP、MQTT、HTTP

• 適用于多系統(tǒng)之間的業(yè)務(wù)解耦的消息中間件

二、基本概念

1、exchange：交換器，負(fù)責(zé)接收消息，轉(zhuǎn)發(fā)消息至綁定的隊(duì)列，有四種類型：

• direct：完全匹配的路由

• topic：模式匹配的路由

• fanout：廣播模式

• headers：鍵值對(duì)匹配路由

Exchange屬性：

• 持久化：如果啟用，那么rabbit服務(wù)重啟之后仍然存在

• 自動(dòng)刪除：如果啟用，那么交換器將會(huì)在其綁定的隊(duì)列都被刪除掉之后自動(dòng)刪除掉自身

2、Queue：隊(duì)列，rabbitmq的內(nèi)部對(duì)象，用于存儲(chǔ)消息，其屬性類似于Exchange，同樣可以設(shè)置是否持久化、自動(dòng)刪除等。

消費(fèi)者從Queue中獲取消息并消費(fèi)。多個(gè)消費(fèi)者可以訂閱同一個(gè)Queue，這時(shí)Queue中的消息會(huì)被平均分?jǐn)偨o多個(gè)消費(fèi)者進(jìn)行處理，而不是每個(gè)消費(fèi)者都收到所有的消息并處理。

3、Binding：綁定，根據(jù)路由規(guī)則綁定交換器與隊(duì)列

4、Routing：路由鍵，路由的關(guān)鍵字

三、消息的可靠性

• Message acknowledgment：消息確認(rèn)，在消息確認(rèn)機(jī)制下，收到回執(zhí)才會(huì)刪除消息，未收到回執(zhí)而斷開(kāi)了連接，消息會(huì)轉(zhuǎn)發(fā)給其他消費(fèi)者，如果忘記回執(zhí)，會(huì)導(dǎo)致消息堆積，消費(fèi)者重啟后會(huì)重復(fù)消費(fèi)這些消息并重復(fù)執(zhí)行業(yè)務(wù)邏輯。

• Message durability：消息持久化，設(shè)置消息持久化可以避免絕大部分消息丟失，比如rabbitmq服務(wù)重啟，但是采用非持久化可以提升隊(duì)列的處理效率。如果要確保消息的持久化，那么消息對(duì)應(yīng)的Exchange和Queue同樣要設(shè)置為持久化。

• Prefetch count，每次發(fā)送給消費(fèi)者消息的數(shù)量，默認(rèn)為1

另外，如果需要可靠性業(yè)務(wù)，需要設(shè)置持久化和ack機(jī)制，如果系統(tǒng)高吞吐，可以設(shè)置為非持久化、noack、自動(dòng)刪除機(jī)制。

四、簡(jiǎn)單應(yīng)用

模擬這樣一個(gè)業(yè)務(wù)場(chǎng)景，用戶下單成功后，需要給用戶增加積分，同時(shí)還需要給用戶發(fā)送下單成功的消息，這是在電商業(yè)務(wù)中很常見(jiàn)的一個(gè)業(yè)務(wù)場(chǎng)景。

如果系統(tǒng)是微服務(wù)架構(gòu)，可能用戶下單功能在訂單服務(wù)，給用戶增加積分的功能在積分服務(wù)，給用戶發(fā)送通知消息的功能在通知服務(wù)，各個(gè)服務(wù)之間解耦，互不影響。那么要實(shí)現(xiàn)上述的業(yè)務(wù)場(chǎng)景，消息中間件rabbitmq是一個(gè)很好的選擇。

原因如下：

• 高性能，它的實(shí)現(xiàn)語(yǔ)言是天生具備高并發(fā)高可用的erlang 語(yǔ)言

• 支持消息的持久化，即使服務(wù)器掛了，也不會(huì)丟失消息

• 消息應(yīng)答（ack）機(jī)制，消費(fèi)者消費(fèi)完消息后發(fā)送一個(gè)消息應(yīng)答，rabbitmq才會(huì)刪除消息，確保消息的可靠性

• 支持高可用集群

• 靈活的路由

實(shí)現(xiàn)思路：

用戶下單成功后，rabbitmq發(fā)送一條消息至EXCHANGE.ORDER_CREATE交換器，該交換器綁定了兩個(gè)隊(duì)列，QUEUE.ORDER_INCREASESCORE、QUEUE.ORDER_NOTIFY，消費(fèi)者訂閱這兩個(gè)隊(duì)列分別用來(lái)處理增加積分、發(fā)送用戶通知。如果后續(xù)日志系統(tǒng)還需要記錄下單的相關(guān)日志，那么我們只需要再定義一個(gè)隊(duì)列并將其綁定到EXCHANGE.ORDER_CREATE即可。

下單發(fā)rabbitmq消息

package?com.robot.rabbitmq;import?com.rabbitmq.client.AMQP; import?com.rabbitmq.client.Channel; import?com.rabbitmq.client.Connection; import?com.rabbitmq.client.ConnectionFactory;import?java.io.IOException; import?java.util.UUID; import?java.util.concurrent.TimeoutException;/***?@author:?會(huì)跳舞的機(jī)器人*?@date:?2017/10/13?10:46*?@description:?模擬用戶下單之后發(fā)送rabbitmq消息*/ public?class?OrderCreator?{//?交換器名稱private?static?final?String?EXCHANGE?=?"EXCHANGE.ORDER_CREATE";//?消息內(nèi)容private?static?String?msg?=?"create?order?success";/***?模擬創(chuàng)建訂單后發(fā)送mq消息*/public?void?createOrder()?{System.out.println("下單成功，開(kāi)始發(fā)送rabbitmq消息");ConnectionFactory?connectionFactory?=?new?ConnectionFactory();connectionFactory.setHost("192.168.12.44");connectionFactory.setPort(56720);connectionFactory.setUsername("baibei");connectionFactory.setPassword("baibei");Connection?connection;Channel?channel;try?{connection?=?connectionFactory.newConnection();channel?=?connection.createChannel();//?持久化boolean?durable?=?true;//?topic類型String?type?=?"topic";//?聲明交換器，如果交換器不存在則創(chuàng)建之channel.exchangeDeclare(EXCHANGE,?type,?durable);String?messgeId?=?UUID.randomUUID().toString();//?deliveryMode>=2表示設(shè)置消息持久化AMQP.BasicProperties?props?=?new?AMQP.BasicProperties.Builder().deliveryMode(2).messageId(messgeId).build();//?發(fā)布消息String?routingKey?=?"order_create";channel.basicPublish(EXCHANGE,?routingKey,?props,?msg.getBytes("utf-8"));connection.close();}?catch?(IOException?e)?{e.printStackTrace();}?catch?(TimeoutException?e)?{e.printStackTrace();}} }

積分系統(tǒng)訂閱消息

package?com.robot.rabbitmq;import?com.rabbitmq.client.AMQP; import?com.rabbitmq.client.Channel; import?com.rabbitmq.client.Connection; import?com.rabbitmq.client.ConnectionFactory; import?com.rabbitmq.client.Consumer; import?com.rabbitmq.client.Envelope; import?com.rabbitmq.client.ShutdownSignalException;import?java.io.IOException; import?java.util.concurrent.TimeoutException;/***?@author:?會(huì)跳舞的機(jī)器人*?@date:?2017/10/13?16:02*?@description:?rabbitmq消費(fèi)者，模擬下單成功后給用戶增加積分*/ public?class?IncreaseScoreConsumer?implements?Consumer?{private?Connection?connection;private?Channel?channel;//?交換器名稱private?static?final?String?EXCHANGE?=?"EXCHANGE.ORDER_CREATE";//?增加積分隊(duì)列名稱private?static?final?String?QUEUENAME?=?"QUEUE.ORDER_INCREASESCORE";public?void?consume()?{//?初始化rabbitmq連接信息ConnectionFactory?connectionFactory?=?new?ConnectionFactory();connectionFactory.setHost("192.168.12.44");connectionFactory.setPort(56720);connectionFactory.setUsername("baibei");connectionFactory.setPassword("baibei");try?{connection?=?connectionFactory.newConnection();channel?=?connection.createChannel();//?聲明交換器channel.exchangeDeclare(EXCHANGE,?"topic",?true);//?聲明隊(duì)列channel.queueDeclare(QUEUENAME,?true,?false,?false,?null);//?交換器與隊(duì)列綁定并設(shè)置routingKeychannel.queueBind(QUEUENAME,?EXCHANGE,?"order_create");//?消費(fèi)消息，callback是該類，關(guān)閉自動(dòng)確認(rèn)消息，在完成業(yè)務(wù)邏輯后手動(dòng)確認(rèn)確認(rèn)channel.basicConsume(QUEUENAME,?false,?this);}?catch?(IOException?e)?{e.printStackTrace();}?catch?(TimeoutException?e)?{e.printStackTrace();}}public?void?handleDelivery(String?consumerTag,?Envelope?envelope,?AMQP.BasicProperties?properties,?byte[]?body)?throws?IOException?{String?msg?=?new?String(body,?"UTF-8");System.out.println("《積分系統(tǒng)》收到訂單消息："?+?msg?+?"，給用戶增加積分......");//?手動(dòng)確認(rèn)消息channel.basicAck(envelope.getDeliveryTag(),?false);/***?channel.basicReject(envelope.getDeliveryTag(),?false);該方法會(huì)丟棄掉隊(duì)列中的這條消息*?channel.basicReject(envelope.getDeliveryTag(),?true);該方法會(huì)把消息重新放回隊(duì)列*?一般系統(tǒng)會(huì)設(shè)定一個(gè)重試次數(shù)，如果超過(guò)重試次數(shù)，則會(huì)丟棄消息，反之則會(huì)把消息再放入隊(duì)列*/}public?void?handleConsumeOk(String?consumerTag)?{}public?void?handleCancelOk(String?consumerTag)?{}public?void?handleCancel(String?consumerTag)?throws?IOException?{}public?void?handleShutdownSignal(String?consumerTag,?ShutdownSignalException?sig)?{}public?void?handleRecoverOk(String?consumerTag)?{}}

通知系統(tǒng)訂閱消息

package?com.robot.rabbitmq;import?com.rabbitmq.client.AMQP; import?com.rabbitmq.client.Channel; import?com.rabbitmq.client.Connection; import?com.rabbitmq.client.ConnectionFactory; import?com.rabbitmq.client.Consumer; import?com.rabbitmq.client.Envelope; import?com.rabbitmq.client.ShutdownSignalException;import?java.io.IOException; import?java.util.concurrent.TimeoutException;/***?@author:?會(huì)跳舞的機(jī)器人*?@date:?2017/10/13?16:20*?@description:?rabbitmq消費(fèi)者，模擬下單成功后給用戶發(fā)送通知*/ public?class?NotifyConsumer?implements?Consumer?{private?Connection?connection;private?Channel?channel;//?交換器名稱private?static?final?String?EXCHANGE?=?"EXCHANGE.ORDER_CREATE";//?通知用戶下單成功通知隊(duì)列名稱private?static?final?String?QUEUENAME?=?"QUEUE.ORDER_NOTIFY";public?void?consume()?{//?初始化rabbitmq連接信息ConnectionFactory?connectionFactory?=?new?ConnectionFactory();connectionFactory.setHost("192.168.12.44");connectionFactory.setPort(56720);connectionFactory.setUsername("baibei");connectionFactory.setPassword("baibei");try?{connection?=?connectionFactory.newConnection();channel?=?connection.createChannel();//?聲明交換器channel.exchangeDeclare(EXCHANGE,?"topic",?true);//?聲明隊(duì)列channel.queueDeclare(QUEUENAME,?true,?false,?false,?null);//?交換器與隊(duì)列綁定并設(shè)置routingKeychannel.queueBind(QUEUENAME,?EXCHANGE,?"order_create");//?消費(fèi)消息，callback是該類，關(guān)閉自動(dòng)確認(rèn)消息，在完成業(yè)務(wù)邏輯后手動(dòng)確認(rèn)確認(rèn)channel.basicConsume(QUEUENAME,?false,?this);}?catch?(IOException?e)?{e.printStackTrace();}?catch?(TimeoutException?e)?{e.printStackTrace();}}public?void?handleDelivery(String?consumerTag,?Envelope?envelope,?AMQP.BasicProperties?properties,?byte[]?body)?throws?IOException?{String?msg?=?new?String(body,?"UTF-8");System.out.println("《通知系統(tǒng)》收到訂單消息："?+?msg?+?"，開(kāi)始給用戶發(fā)送通知......");//?手動(dòng)確認(rèn)消息channel.basicAck(envelope.getDeliveryTag(),?false);/***?channel.basicReject(envelope.getDeliveryTag(),?false);該方法會(huì)丟棄掉隊(duì)列中的這條消息*?channel.basicReject(envelope.getDeliveryTag(),?true);該方法會(huì)把消息重新放回隊(duì)列*?一般系統(tǒng)會(huì)設(shè)定一個(gè)重試次數(shù)，如果超過(guò)重試次數(shù)，則會(huì)丟棄消息，反之則會(huì)把消息再放入隊(duì)列*/}public?void?handleConsumeOk(String?consumerTag)?{}public?void?handleCancelOk(String?consumerTag)?{}public?void?handleCancel(String?consumerTag)?throws?IOException?{}public?void?handleShutdownSignal(String?consumerTag,?ShutdownSignalException?sig)?{}public?void?handleRecoverOk(String?consumerTag)?{} }

測(cè)試

package?com.robot.rabbitmq;/***?@author:?會(huì)跳舞的機(jī)器人*?@date:?2017/10/13?16:27*?@description:*/ public?class?Test?{public?static?void?main(String[]?args)?{IncreaseScoreConsumer?increaseScoreConsumer?=?new?IncreaseScoreConsumer();increaseScoreConsumer.consume();NotifyConsumer?notifyConsumer?=?new?NotifyConsumer();notifyConsumer.consume();OrderCreator?orderCreator?=?new?OrderCreator();for?(int?i?=?0;?i?<?3;?i++)?{orderCreator.createOrder();}} }

輸出：

下單成功，開(kāi)始發(fā)送rabbitmq消息 《積分系統(tǒng)》收到訂單消息：create order success，給用戶增加積分...... 《通知系統(tǒng)》收到訂單消息：create order success，開(kāi)始給用戶發(fā)送通知...... 下單成功，開(kāi)始發(fā)送rabbitmq消息 《積分系統(tǒng)》收到訂單消息：create order success，給用戶增加積分...... 《通知系統(tǒng)》收到訂單消息：create order success，開(kāi)始給用戶發(fā)送通知...... 下單成功，開(kāi)始發(fā)送rabbitmq消息 《積分系統(tǒng)》收到訂單消息：create order success，給用戶增加積分...... 《通知系統(tǒng)》收到訂單消息：create order success，開(kāi)始給用戶發(fā)送通知...

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ简介以及应用的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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