生產(chǎn)消息 

在本配方中, 我們將學(xué)習(xí)了如何將消息發(fā)送到AMQP隊(duì)列. 我們將介紹AMQP消息的構(gòu)建塊:消息,隊(duì)列,以及交換器.你可以在Chapter01/Recipe02/src/rmqexample中找到代碼.
w to do it…
在連接到中間件后, 像前面配方中看到的一樣,你可以按下面的步驟來來發(fā)送消息:
1. 聲明隊(duì)列, 在 com.rabbitmq.client.Channel上調(diào)用queueDeclare()方法:
String myQueue = "myFirstQueue";
channel.queueDeclare(myQueue, true, false, false, null); //創(chuàng)建一個(gè)名為myFirstQueue,持久化的,非限制的,不自動(dòng)刪除的隊(duì)列,
2. 發(fā)送第一個(gè)消息到RabbitMQ broker:
String message = "My message to myFirstQueue";
channel.basicPublish("",myQueue, null, message.getBytes());
3. 使用不同的選項(xiàng)發(fā)送第二個(gè)消息:
channel.basicPublish("",myQueue,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

注意:隊(duì)列名稱是大小寫敏感的: MYFIRSTQUEUE與myFirstQueue是不同的.
如何工作

public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws java.io.IOException {

String messageIdentifier =properties.getCorrelationId();

String action = actions.get(messageIdentifier);

actions.remove(messageIdentifier);

String response = new String(body);

OnReply(action, response);

}

4. 調(diào)用channel.basicConsume()方法啟動(dòng)消息消費(fèi).

5. 準(zhǔn)備和序列化請(qǐng)求(我們例子中是messageRequest).

6. 初始化一個(gè)任意的唯一消息標(biāo)識(shí)符(messageIdentifier).

7. 當(dāng)消費(fèi)者收到相應(yīng)回復(fù)的時(shí)候，定義應(yīng)該做什么，通過使用messageIdentifier來綁定動(dòng)作.在我們的例子中，我們通過調(diào)用我們自定義的方法 consumer.AddAction()來完成的.

8. 發(fā)布消息到請(qǐng)求隊(duì)列,設(shè)置其屬性:

BasicProperties props = new BasicProperties.Builder().correlationId(messageIdentifier)replyTo(replyQueue).build();

channel.basicPublish("", requestQueue,props,messageRequest.getBytes());

如何工作

在我們的例子中，RPC responder扮演的是RPC server的角色; responder會(huì)監(jiān)聽requestQueue公共隊(duì)列(步驟1),這里放置了調(diào)用者的請(qǐng)求.

另一方面，每個(gè)調(diào)用者會(huì)在其私有隊(duì)列上消費(fèi)responder的回復(fù)信息(步驟5).當(dāng)caller發(fā)送消息時(shí)(步驟11),它包含兩個(gè)屬性:一個(gè)是用于監(jiān)聽的臨時(shí)回復(fù)隊(duì)列 (replyTo())名稱,另一個(gè)是消息標(biāo)識(shí)(correlationId()),當(dāng)回復(fù)消息時(shí)，用于標(biāo)識(shí)caller.事實(shí)上,在我們的例子中，我們已經(jīng)實(shí)現(xiàn)了一個(gè)異步的RPC caller. The action to be performed by the RpcCallerConsumer (step 6) when the reply comes back is recorded by the nonblocking consumer by calling AddAction() (step 10).