? ? ?本文是簡單介紹一下RabbitMQ，參考官網上的教程。同時加入了一些自己的理解。官網教程詳見：["Hello World!"](http://www.rabbitmq.com/tutorials/tutorial-one-java.html)。 # 引言 ? ? ?你是否遇到過多個系統間需要通過定時任務來同步某些數據？ ? ? ?你是否在為異構系統的不同進程間相互調用、通訊的問題而苦惱、掙扎？ ? ? ?如果是，那么恭喜你，消息服務讓你可以很輕松地解決這些問題。消息服務擅長于解決多系統、異構系統間的數據交換（消息通知/通訊）問題。 ? ? ?本文將要介紹的RabbitMQ就是當前最主流的消息中間件之一。 # RabbitMQ簡介 ? ? ?MQ（Message Queue，消息隊列）是一種應用系統之間的通信方法。是通過讀寫出入隊列的消息來通信（RPC則是通過直接調用彼此來通信的）。 ? ? ?AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標準，為面向消息的中間件設計。消息中間件主要用于組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/訂閱）、可靠性、安全。 ? ? ?RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。 ? ? ?下面通過生產者代碼來解釋一下RabbitMQ中涉及到的概念。 ~~~ public class MsgSender { private final static String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException { /** * 創建連接連接到MabbitMQ */ ConnectionFactory factory = new ConnectionFactory(); // 設置MabbitMQ所在主機ip或者主機名 factory.setHost("127.0.0.1"); // 創建一個連接 Connection connection = factory.newConnection(); // 創建一個頻道 Channel channel = connection.createChannel(); // 指定一個隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 發送的消息 String message = "hello world!龍軒"; // 往隊列中發出一條消息 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); // 關閉頻道和連接 channel.close(); connection.close(); } } ~~~ ### ConnectionFactory、Connection、Channel ? ? ?ConnectionFactory、Connection、Channel，這三個都是RabbitMQ對外提供的API中最基本的對象。不管是服務器端還是客戶端都會首先創建這三類對象。 ? ? ?ConnectionFactory為Connection的制造工廠。 ? ? ?Connection是與RabbitMQ服務器的socket鏈接，它封裝了socket協議及身份驗證相關部分邏輯。 ? ? ?Channel是我們與RabbitMQ打交道的最重要的一個接口，大部分的業務操作是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、發布消息等。 ### Queue ? ? ?Queue（隊列）是RabbitMQ的內部對象，用于存儲消息，用下圖表示。  ? ? ?RabbitMQ中的消息都只能存儲在Queue中，生產者（下圖中的P）生產消息并最終投遞到Queue中，消費者（下圖中的C）可以從Queue中獲取消息并消費。  ? ? ?隊列是有Channel聲明的，而且這個操作是[冪等](http://baike.baidu.com/view/2067025.htm)的。同名的隊列多次聲明也只會創建一次。我們發送消息就是想這個聲明的隊列里發送消息。 ? ? ?看一下消費者的代碼： ~~~ public class MsgReceiver { private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws IOException, InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); // 打開連接和創建頻道，與發送端一樣 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 聲明隊列，主要為了防止消息接收者先運行此程序，隊列還不存在時創建隊列。 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 創建隊列消費者 QueueingConsumer consumer = new QueueingConsumer(channel); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 指定消費隊列 channel.basicConsume(QUEUE_NAME, true, consumer); while (true) { // nextDelivery是一個阻塞方法（內部實現其實是阻塞隊列的take方法） QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); } } } ~~~ ? ? ?從上述代碼中，我們可以看到ConnectionFactory、Connection、Channel這三個對象都還是會創建。而隊列在消費者這里又聲明了一遍。這是為了防止先啟動消費者，當為消費者指定隊列時，如果RabbitMQ服務器上未聲明過隊列，就會拋出IO異常。 ### QueueingConsumer ? ? ?隊列消費者，用于監聽隊列中的消息。調用nextDelivery方法時，內部實現就是調用隊列的take方法。該方法的作用：獲取并移除此隊列的頭部，在元素變得可用之前一直等待（如果有必要）。說白了就是如果沒有消息，就處于阻塞狀態。 ? ? ?運行結果如下：（生產者、消費者誰先運行都可以）