轉載：輕松搞定RabbitMQ
一、RabbitMQ
AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標准，為面向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。
AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/訂閱）、可靠性、安全。
RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用於在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。
二、RabbitMQ的使用場景
對於一個大型的軟件系統來說，它會有很多的組件或者說模塊或者說子系統或者（subsystem or Component or submodule）。那么這些模塊的如何通信？這和傳統的IPC有很大的區別。傳統的IPC很多都是在單一系統上的，模塊耦合性很大，不適合擴展（Scalability）；如果使用socket那么不同的模塊的確可以部署到不同的機器上，但是還是有很多問題需要解決。比如：
1）信息的發送者和接收者如何維持這個連接，如果一方的連接中斷，這期間的數據如何方式丟失？
2）如何降低發送者和接收者的耦合度？
3）如何讓Priority高的接收者先接到數據？
4）如何做到load balance？有效均衡接收者的負載？
5）如何有效的將數據發送到相關的接收者？也就是說將接收者subscribe 不同的數據，如何做有效的filter。
6）如何做到可擴展，甚至將這個通信模塊發到cluster上？
7）如何保證接收者接收到了完整，正確的數據？
AMDQ協議解決了以上的問題，而RabbitMQ實現了AMQP。
三、RabbitMQ的結構
RabbitMQ的應用場景架構圖如下：
- Broker：簡單來說就是消息隊列服務器實體。
- Exchange：消息交換機，它指定消息按什么規則，路由到哪個隊列。
- Queue：消息隊列載體，每個消息都會被投入到一個或多個隊列。
- Binding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。
- Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。
- vhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。
- producer：消息生產者，就是投遞消息的程序。
- consumer：消息消費者，就是接受消息的程序。
- channel：消息通道，在客戶端的每個連接里，可建立多個channel，每個channel代表一個會話任務。
四、RabbitMQ的使用過程
AMQP模型中，消息在producer中產生，發送到MQ的exchange上，exchange根據配置的路由方式發到相應的Queue上，Queue又將消息發送給consumer，消息從queue到consumer有push和pull兩種方式。 消息隊列的使用過程大概如下：
- 客戶端連接到消息隊列服務器，打開一個channel。
- 客戶端聲明一個exchange，並設置相關屬性。
- 客戶端聲明一個queue，並設置相關屬性。
- 客戶端使用routing key，在exchange和queue之間建立好綁定關系。
- 客戶端投遞消息到exchange。
4.0 安裝和配置
RabbitMQ使用Erlang語言實現，因此在使用時首先要安裝和配置erlang環境，並安裝服務器后進行相關配置，由於不是本文主要內容所以忽略，詳見RabbitMQ簡介。
RabbitMQ的客戶端使用時需要添加相關依賴。
4.1 點對點
消息生產者的代碼如下：
-
package com.zenhobby.rabbit.demo;
-
-
import com.rabbitmq.client.Channel;
-
import com.rabbitmq.client.Connection;
-
import com.rabbitmq.client.ConnectionFactory;
-
-
public
class Send
-
{
-
//隊列名稱
-
private
final
static String QUEUE_NAME =
"hello";
-
-
public static void main(String[] argv) throws java.io.IOException
-
{
-
/**
-
* 創建連接連接到MabbitMQ
-
*/
-
ConnectionFactory factory =
new ConnectionFactory();
-
//設置MabbitMQ所在主機ip或者主機名
-
factory.setHost(
"localhost");
-
//創建一個連接
-
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();
-
}
-
}
-
package com.zenhobby.rabbit.demo;
-
-
import com.rabbitmq.client.Channel;
-
import com.rabbitmq.client.Connection;
-
import com.rabbitmq.client.ConnectionFactory;
-
import com.rabbitmq.client.QueueingConsumer;
-
-
public
class Recv
-
{
-
//隊列名稱
-
private
final
static String QUEUE_NAME =
"hello";
-
-
public static void main(String[] argv) throws java.io.IOException,
-
java.lang.InterruptedException
-
{
-
//打開連接和創建頻道，與發送端一樣
-
ConnectionFactory factory =
new ConnectionFactory();
-
factory.setHost(
"localhost");
-
Connection connection = factory.newConnection();
-
Channel channel = connection.createChannel();
-
//聲明隊列，主要為了防止消息接收者先運行此程序，隊列還不存在時創建隊列。
-
channel.queueDeclare(QUEUE_NAME,
false,
false,
false,
null);
-
System.out.println(
" [*] Waiting for messages. To exit press CTRL+C");
-
-
//創建隊列消費者
-
QueueingConsumer consumer =
new QueueingConsumer(channel);
-
//指定消費隊列，關閉默認的消息應答
-
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 +
"'");
-
}
-
-
}
-
}
4.2 工作隊列
工作隊列的主要任務是：避免立刻執行資源密集型任務，然后必須等待其完成。相反地，我們進行任務調度：我們把任務封裝為消息發送給隊列。工作進行在后台運行並不斷的從隊列中取出任務然后執行。當你運行了多個工作進程時，任務隊列中的任務將會被工作進程共享執行。這樣的概念在web應用中極其有用，當在很短的HTTP請求間需要執行復雜的任務。
1.消息分發機制
默認的，RabbitMQ會一個一個的發送信息給下一個消費者(consumer)，而不考慮每個任務的時長等等，且是一次性分配，並非一個一個分配。平均的每個消費者將會獲得相等數量的消息。這樣分發消息的方式叫做round-robin。
默認的任務分發雖然看似公平但存在弊端。比如：現在有2個消費者，所有的奇數的消息都是繁忙的，而偶數則是輕松的。按照輪詢的方式，奇數的任務交給了第一個消費者，所以一直在忙個不停。偶數的任務交給另一個消費者，則立即完成任務，然后閑得不行。而RabbitMQ則是不了解這些的。這是因為當消息進入隊列，RabbitMQ就會分派消息。它不看消費者為應答的數目，只是盲目的將第n條消息發給第n個消費者。
為了解決這個問題，我們使用basicQos( prefetchCount = 1)方法，來限制RabbitMQ只發不超過1條的消息給同一個消費者。當消息處理完畢后，有了反饋，才會進行第二次發送。
-
int prefetchCount =
1;
-
channel.basicQos(prefetchCount);
2. 消息確認
每個Consumer可能需要一段時間才能處理完收到的數據。如果在這個過程中，Consumer出錯了，異常退出了，而數據還沒有處理完成，那么非常不幸，這段數據就丟失了。因為我們采用no-ack的方式進行確認，也就是說，每次Consumer接到數據后，而不管是否處理完成，RabbitMQ Server會立即把這個Message標記為完成，然后從queue中刪除了。
為了保證數據不被丟失，RabbitMQ支持消息確認機制，即acknowledgments。為了保證數據能被正確處理而不僅僅是被Consumer收到，那么我們不能采用no-ack。而應該是在處理完數據后發送ack。在處理數據后發送的ack，就是告訴RabbitMQ數據已經被接收，處理完成，RabbitMQ可以去安全的刪除它了。如果Consumer退出了但是沒有發送ack，那么RabbitMQ就會把這個Message發送到下一個Consumer。這樣就保證了在Consumer異常退出的情況下數據也不會丟失。這里並沒有用到超時機制。RabbitMQ僅僅通過Consumer的連接中斷來確認該Message並沒有被正確處理。也就是說，RabbitMQ給了Consumer足夠長的時間來做數據處理。
默認情況下，消息確認是打開的（enabled）：
-
boolean autoAck =
false;
-
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
-
channel.basicQos(
1);
//保證一次只分發一個
-
// 創建隊列消費者
-
final Consumer consumer =
new DefaultConsumer(channel) {
-
@Override
-
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
-
String message =
new String(body,
"UTF-8");
-
-
System.out.println(
" [x] Received '" + message +
"'");
-
System.out.println(
" [x] Proccessing... at " +
new Date().toLocaleString());
-
try {
-
for (
char ch: message.toCharArray()) {
-
if (ch ==
'.') {
-
Thread.sleep(
1000);
-
}
-
}
-
}
catch (InterruptedException e) {
-
}
finally {
-
System.out.println(
" [x] Done! at " +
new Date().toLocaleString());
-
channel.basicAck(envelope.getDeliveryTag(),
false);
-
}
-
}
-
};
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
3. 消息持久化
為了保證在RabbitMQ退出或者crash了數據仍沒有丟失，需要將queue和Message都要持久化。queue的持久化需要在聲明時指定durable=True：
channel.queue_declare(queue='hello', durable=True)
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
修改后的生產者如下所示：
-
static void Main(string[] args)
-
{
-
var factory =
new ConnectionFactory() { HostName =
"localhost" };
-
using (var connection = factory.CreateConnection())
-
{
-
using (var channel = connection.CreateModel())
-
{
-
bool durable =
true;
-
channel.QueueDeclare(
"task_queue", durable,
false,
false,
null);
//queue的持久化需要在聲明時指定durable=True
-
var message = GetMessage(args);
-
var body = Encoding.UTF8.GetBytes(message);
-
var properties = channel.CreateBasicProperties();
-
properties.SetPersistent(
true);
//需要持久化Message，即在Publish的時候指定一個properties，
-
channel.BasicPublish(
"",
"task_hello", properties, body);
-
}
-
}
-
}
4.3 Publish/Subscribe
1. 交換器
在工作隊列一節中使用的分發如下：
channel.basicPublish("", "hello", null, message.getBytes());
交換器的規則有：
- direct （直連）：
- topic （主題）
- headers （標題）
- fanout （分發）
Direct Exchange – 處理路由鍵。需要將一個隊列綁定到交換機上，要求該消息與一個特定的路由鍵完全匹配。這是一個完整的匹配。如果一個隊列綁定到該交換機上要求路由鍵 “dog”，則只有被標記為“dog”的消息才被轉發，不會轉發dog.puppy，也不會轉發dog.guard，只會轉發dog。
-
Channel channel = connection.createChannel();
-
channel.exchangeDeclare(
"exchangeName",
"direct");
//direct fanout topic
-
channel.queueDeclare(
"queueName");
-
channel.queueBind(
"queueName",
"exchangeName",
"routingKey");
-
-
byte[] messageBodyBytes =
"hello world".getBytes();
-
//需要綁定路由鍵
-
channel.basicPublish(
"exchangeName",
"routingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
Fanout Exchange – 不處理路由鍵。你只需要簡單的將隊列綁定到交換機上。一個發送到交換機的消息都會被轉發到與該交換機綁定的所有隊列上。很像子網廣播，每台子網內的主機都獲得了一份復制的消息。 Fanout交換機轉發消息是最快的。
-
Channel channel = connection.createChannel();
-
channel.exchangeDeclare(
"exchangeName",
"fanout");
//direct fanout topic
-
channel.queueDeclare(
"queueName");
-
channel.queueBind(
"queueName",
"exchangeName",
"routingKey");
-
-
channel.queueDeclare(
"queueName1");
-
channel.queueBind(
"queueName1",
"exchangeName",
"routingKey1");
-
-
byte[] messageBodyBytes =
"hello world".getBytes();
-
//路由鍵需要設置為空
-
channel.basicPublish(
"exchangeName",
"", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
Topic Exchange – 將路由鍵和某模式進行匹配。此時隊列需要綁定要一個模式上。符號“#”匹配一個或多個詞，符號“*”匹配不多不少一個詞。因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*” 只會匹配到“audit.irs”。
-
Channel channel = connection.createChannel();
-
channel.exchangeDeclare(
"exchangeName",
"topic");
//direct fanout topic
-
channel.queueDeclare(
"queueName");
-
channel.queueBind(
"queueName",
"exchangeName",
"routingKey.*");
-
-
byte[] messageBodyBytes =
"hello world".getBytes();
-
channel.basicPublish(
"exchangeName",
"routingKey.one", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
Header Exchange
Headers類型的exchange使用的比較少，它也是忽略routingKey的一種路由方式。是使用Headers來匹配的。
Headers是一個鍵值對，可以定義成Hashtable。發送者在發送的時候定義一些鍵值對，接收者也可以再綁定時候傳入一些鍵值對，兩者匹配的話，則對應的隊列就可以收到消息。匹配有兩種方式all和any。這兩種方式是在接收端必須要用鍵值"x-mactch"來定義。
all代表定義的多個鍵值對都要滿足，而any則代碼只要滿足一個就可以了。
fanout，direct，topic exchange的routingKey都需要要字符串形式的，而headers exchange則沒有這個要求，因為鍵值對的值可以是任何類型。
消息生產者如下：
-
package cn.slimsmart.rabbitmq.demo.headers;
-
-
import java.util.Date;
-
import java.util.Hashtable;
-
import java.util.Map;
-
-
import org.springframework.amqp.core.ExchangeTypes;
-
-
import com.rabbitmq.client.AMQP;
-
import com.rabbitmq.client.AMQP.BasicProperties;
-
import com.rabbitmq.client.AMQP.BasicProperties.Builder;
-
import com.rabbitmq.client.Channel;
-
import com.rabbitmq.client.Connection;
-
import com.rabbitmq.client.ConnectionFactory;
-
-
public
class Producer {
-
private
final
static String EXCHANGE_NAME =
"header-exchange";
-
-
@SuppressWarnings(
"deprecation")
-
public static void main(String[] args) throws Exception {
-
// 創建連接和頻道
-
ConnectionFactory factory =
new ConnectionFactory();
-
factory.setHost(
"192.168.36.102");
-
// 指定用戶 密碼
-
factory.setUsername(
"admin");
-
factory.setPassword(
"admin");
-
// 指定端口
-
factory.setPort(AMQP.PROTOCOL.PORT);
-
Connection connection = factory.newConnection();
-
Channel channel = connection.createChannel();
-
-
//聲明轉發器和類型headers
-
channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,
false,
true,
null);
-
String message =
new Date().toLocaleString() +
" : log something";
-
-
Map<String,Object> headers =
new Hashtable<String, Object>();
-
headers.put(
"aaa",
"01234");
-
Builder properties =
new BasicProperties.Builder();
-
properties.headers(headers);
-
-
// 指定消息發送到的轉發器,綁定鍵值對headers鍵值對
-
channel.basicPublish(EXCHANGE_NAME,
"",properties.build(),message.getBytes());
-
-
System.out.println(
"Sent message :'" + message +
"'");
-
channel.close();
-
connection.close();
-
}
-
}
消息消費者如下：
-
package cn.slimsmart.rabbitmq.demo.headers;
-
-
import java.util.Hashtable;
-
import java.util.Map;
-
-
import org.springframework.amqp.core.ExchangeTypes;
-
-
import com.rabbitmq.client.AMQP;
-
import com.rabbitmq.client.Channel;
-
import com.rabbitmq.client.Connection;
-
import com.rabbitmq.client.ConnectionFactory;
-
import com.rabbitmq.client.QueueingConsumer;
-
-
public
class Consumer {
-
private
final
static String EXCHANGE_NAME =
"header-exchange";
-
private
final
static String QUEUE_NAME =
"header-queue";
-
-
public static void main(String[] args) throws Exception {
-
// 創建連接和頻道
-
ConnectionFactory factory =
new ConnectionFactory();
-
factory.setHost(
"192.168.36.102");
-
// 指定用戶 密碼
-
factory.setUsername(
"admin");
-
factory.setPassword(
"admin");
-
// 指定端口
-
factory.setPort(AMQP.PROTOCOL.PORT);
-
Connection connection = factory.newConnection();
-
Channel channel = connection.createChannel();
-
-
//聲明轉發器和類型headers
-
channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,
false,
true,
null);
-
channel.queueDeclare(QUEUE_NAME,
false,
false,
true,
null);
-
-
Map<String, Object> headers =
new Hashtable<String, Object>();
-
headers.put(
"x-match",
"any");
//all any
-
headers.put(
"aaa",
"01234");
-
headers.put(
"bbb",
"56789");
-
// 為轉發器指定隊列，設置binding 綁定header鍵值對
-
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME,
"", headers);
-
QueueingConsumer consumer =
new QueueingConsumer(channel);
-
// 指定接收者，第二個參數為自動應答，無需手動應答
-
channel.basicConsume(QUEUE_NAME,
true, consumer);
-
while (
true) {
-
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
-
String message =
new String(delivery.getBody());
-
System.out.println(message);
-
}
-
}
-
}
Default Exchange
其實除了上面四種以外還有一種Default Exchange，它是一種特別的Direct Exchange。
當你手動創建一個隊列時，后台會自動將這個隊列綁定到一個名稱為空的Direct類型交換機上，綁定路由名稱與隊列名稱相同。有了這個默認的交換機和綁定，我們就可以像其他輕量級的隊列，如Redis那樣，直接操作隊列來處理消息。不過只是看起來是，實際上在RabbitMQ里直接操作是不可能的。消息始終都是先發送到交換機，由交換級經過路由傳送給隊列，消費者再從隊列中獲取消息的。不過由於這個默認交換機和路由的關系，使我們只關心隊列這一層即可，這個比較適合做一些簡單的應用，畢竟沒有發揮RabbitMQ的最大功能，如果都用這種方式去使用的話就真是殺雞用宰牛刀了。
2. 臨時隊列
如果要在生產者和消費者之間創建一個新的隊列，又不想使用原來的隊列，臨時隊列就是為這個場景而生的：
首先，每當我們連接到RabbitMQ，我們需要一個新的空隊列，我們可以用一個隨機名稱來創建，或者說讓服務器選擇一個隨機隊列名稱給我們。
一旦我們斷開消費者，隊列應該立即被刪除。Java客戶端提供queuedeclare()為我們創建一個非持久化、獨立、自動刪除的隊列名稱。
String queueName = channel.queueDeclare().getQueue();
3. 綁定
如果我們已經創建了一個分發交換器和隊列，現在我們就可以就將我們的隊列跟交換器進行綁定。
channel.queueBind(queueName, "logs", "");
五、RabbitMQ實現RPC
RabbitMQ可以用於實現RPC，兩者有相像之處，使用RabbitMQ實現RPC分為如下幾個步驟：
1. Client interface（客戶端接口）
為了說明RPC服務可以使用，我們創建一個簡單的客戶端類。暴露一個方法——發送RPC請求，然后阻塞直到獲得結果。
-
FibonacciRpcClient fibonacciRpc =
new FibonacciRpcClient();
-
String result = fibonacciRpc.call(
"4");
-
System.out.println(
"fib(4) is " + result);
2. Callback queue（回調隊列）
一般在RabbitMQ中做RPC是很簡單的。客戶端發送請求消息，服務器回復響應的消息。為了接受響應的消息，我們需要在請求消息中發送一個回調隊列。可以用默認的隊列：
-
BasicProperties props =
new BasicProperties
-
.Builder()
-
.replyTo(callbackQueueName)
-
.build();
-
-
channel.basicPublish(
"",
"rpc_queue", props, message.getBytes());
-
-
// ... then code to read a response message from the callback_queue ...
3. Message properties（消息屬性）
AMQP協議為消息預定義了一組14個屬性。大部分的屬性是很少使用的。除了一下幾種：
- deliveryMode：標記消息傳遞模式，2-消息持久化，其他值-瞬態。在第二篇文章中還提到過。
- contentType：內容類型，用於描述編碼的mime-type。例如經常為該屬性設置JSON編碼。
- replyTo：應答，通用的回調隊列名稱
- correlationId：關聯ID，方便RPC響應與請求關聯
import com.rabbitmq.client.AMQP.BasicProperties;
4. Correlation Id
在上述方法中為每個RPC請求創建一個回調隊列。這是很低效的。幸運的是，一個解決方案：可以為每個客戶端創建一個單一的回調隊列。
新的問題被提出，隊列收到一條回復消息，但是不清楚是那條請求的回復。這是就需要使用correlationId屬性了。我們要為每個請求設置唯一的值。然后，在回調隊列中獲取消息，看看這個屬性，關聯response和request就是基於這個屬性值的。如果我們看到一個未知的correlationId屬性值的消息，可以放心的無視它——它不是我們發送的請求。
你可能問道，為什么要忽略回調隊列中未知的信息，而不是當作一個失敗？這是由於在服務器端競爭條件的導致的。雖然不太可能，但是如果RPC服務器在發送給我們結果后，發送請求反饋前就掛掉了，這有可能會發送未知correlationId屬性值的消息。如果發生了這種情況，重啟RPC服務器將會重新處理該請求。這就是為什么在客戶端必須很好的處理重復響應，RPC應該是冪等的。
5. 實現
我們的RPC的處理流程：
- 當客戶端啟動時，創建一個匿名的回調隊列。
- 客戶端為RPC請求設置2個屬性：replyTo：設置回調隊列名字；correlationId：標記request。
- 請求被發送到rpc_queue隊列中。
- RPC服務器端監聽rpc_queue隊列中的請求，當請求到來時，服務器端會處理並且把帶有結果的消息發送給客戶端。接收的隊列就是replyTo設定的回調隊列。
- 客戶端監聽回調隊列，當有消息時，檢查correlationId屬性，如果與request中匹配，那就是結果了。
RPC服務器端（RPCServer.java）
-
/**
-
* RPC服務器端
-
*
-
* @author arron
-
* @date 2015年9月30日 下午3:49:01
-
* @version 1.0
-
*/
-
public
class RPCServer {
-
-
private
static
final String RPC_QUEUE_NAME =
"rpc_queue";
-
-
public static void main(String[] args) throws Exception {
-
ConnectionFactory factory =
new ConnectionFactory();
-
// 設置MabbitMQ所在主機ip或者主機名
-
factory.setHost(
"127.0.0.1");
-
// 創建一個連接
-
Connection connection = factory.newConnection();
-
// 創建一個頻道
-
Channel channel = connection.createChannel();
-
-
//聲明隊列
-
channel.queueDeclare(RPC_QUEUE_NAME,
false,
false,
false,
null);
-
-
//限制：每次最多給一個消費者發送1條消息
-
channel.basicQos(
1);
-
-
//為rpc_queue隊列創建消費者，用於處理請求
-
QueueingConsumer consumer =
new QueueingConsumer(channel);
-
channel.basicConsume(RPC_QUEUE_NAME,
false, consumer);
-
-
System.out.println(
" [x] Awaiting RPC requests");
-
-
while (
true) {
-
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
-
-
//獲取請求中的correlationId屬性值，並將其設置到結果消息的correlationId屬性中
-
BasicProperties props = delivery.getProperties();
-
BasicProperties replyProps =
new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
-
//獲取回調隊列名字
-
String callQueueName = props.getReplyTo();
-
-
String message =
new String(delivery.getBody(),
"UTF-8");
-
-
System.out.println(
" [.] fib(" + message +
")");
-
-
//獲取結果
-
String response =
"" + fib(Integer.parseInt(message));
-
//先發送回調結果
-
channel.basicPublish(
"", callQueueName, replyProps,response.getBytes());
-
//后手動發送消息反饋
-
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),
false);
-
}
-
}
-
-
/**
-
* 計算斐波列其數列的第n項
-
*
-
* @param n
-
* @return
-
* @throws Exception
-
*/
-
private static int fib(int n) throws Exception {
-
if (n <
0)
-
throw
new Exception(
"參數錯誤，n必須大於等於0");
-
if (n ==
0)
-
return
0;
-
if (n ==
1)
-
return
1;
-
return fib(n -
1) + fib(n -
2);
-
}
-
}
-
/**
-
*
-
* @author arron
-
* @date 2015年9月30日 下午3:44:43
-
* @version 1.0
-
*/
-
public
class RPCClient {
-
-
private
static
final String RPC_QUEUE_NAME =
"rpc_queue";
-
-
private Connection connection;
-
private Channel channel;
-
private String replyQueueName;
-
private QueueingConsumer consumer;
-
-
public RPCClient() throws Exception {
-
ConnectionFactory factory =
new ConnectionFactory();
-
// 設置MabbitMQ所在主機ip或者主機名
-
factory.setHost(
"127.0.0.1");
-
// 創建一個連接
-
connection = factory.newConnection();
-
// 創建一個頻道
-
channel = connection.createChannel();
-
-
//聲明隊列
-
channel.queueDeclare(RPC_QUEUE_NAME,
false,
false,
false,
null);
-
-
//為每一個客戶端獲取一個隨機的回調隊列
-
replyQueueName = channel.queueDeclare().getQueue();
-
//為每一個客戶端創建一個消費者（用於監聽回調隊列，獲取結果）
-
consumer =
new QueueingConsumer(channel);
-
//消費者與隊列關聯
-
channel.basicConsume(replyQueueName,
true, consumer);
-
}
-
-
/**
-
* 獲取斐波列其數列的值
-
*
-
* @param message
-
* @return
-
* @throws Exception
-
*/
-
public String call(String message) throws Exception{
-
String response =
null;
-
String corrId = java.util.UUID.randomUUID().toString();
-
-
//設置replyTo和correlationId屬性值
-
BasicProperties props =
new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();
-
-
//發送消息到rpc_queue隊列
-
channel.basicPublish(
"", RPC_QUEUE_NAME, props, message.getBytes());
-
-
while (
true) {
-
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
-
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
-
response =
new String(delivery.getBody(),
"UTF-8");
-
break;
-
}
-
}
-
-
return response;
-
}
-
-
public static void main(String[] args) throws Exception {
-
RPCClient fibonacciRpc =
new RPCClient();
-
String result = fibonacciRpc.call(
"4");
-
System.out.println(
"fib(4) is " + result);
-
}
-
}
這里的例子只是RabbitMQ中RPC服務的一個實現，你也可以根據業務需要實現更多。rpc有一個優點，如果一個RPC服務器處理不來，可以再增加一個、兩個、三個。我們的例子中的代碼還比較簡單，還有很多問題沒有解決：
- 如果沒有發現服務器，客戶端如何處理？
- 如果客戶端的RPC請求超時了怎么辦？
- 如果服務器出現了故障，發生了異常，是否將異常發送到客戶端？
- 在處理消息前，怎樣防止無效的消息？檢查范圍、類型？
以上內容轉載自網絡，詳見開頭轉載聲明。