RabbitMQ Java client 將com.rabbitmq.client作為其頂層包. 關(guān)鍵類和接口有:

• Channel
• Connection
• ConnectionFactory
• Consumer

Connections 和 Channels

核心API類是Connection和Channel, 它們代表對應(yīng)AMQP 0-9-1 connection 和 channel. 在使用前，可像下面這樣來導(dǎo)入：

import com.rabbitmq.client.Connection; 
import com.rabbitmq.client.Channel;

連接到broker

下面的代碼會(huì)使用給定的參數(shù)連接到AMQP broker:

ConnectionFactory factory = new ConnectionFactory(); 
factory.setUsername(userName); 
factory.setPassword(password); 
factory.setVirtualHost(virtualHost); 
factory.setHost(hostName); 
factory.setPort(portNumber); 
Connection conn = factory.newConnection(); 

也可以使用URIs 來設(shè)置連接參數(shù):

ConnectionFactory factory = new ConnectionFactory(); 
factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost"); 
Connection conn = factory.newConnection(); 

Connection 接口可用來打開一個(gè)channel:

Channel channel = conn.createChannel(); 

channel現(xiàn)在可用來發(fā)送和接收消息，正如后續(xù)章節(jié)中描述的一樣.

要斷開連接，只需要簡單地關(guān)閉channel和connection:

channel.close(); conn.close();

關(guān)閉channel被認(rèn)為是最佳實(shí)踐,但在這里不是嚴(yán)格必須的 - 當(dāng)?shù)讓舆B接關(guān)閉的時(shí)候，channel也會(huì)自動(dòng)關(guān)閉.

使用 Exchanges 和 Queues

采用交換器和隊(duì)列工作的客戶端應(yīng)用程序,是AMQP高級別構(gòu)建模塊。在使用前，必須先聲明.聲明每種類型的對象都需要確保名稱存在，如果有必要須進(jìn)行創(chuàng)建.

繼續(xù)上面的例子,下面的代碼聲明了一個(gè)交換器和一個(gè)隊(duì)列，然后再將它們進(jìn)行綁定.

channel.exchangeDeclare(exchangeName, "direct", true); 
String queueName = channel.queueDeclare().getQueue(); 
channel.queueBind(queueName, exchangeName, routingKey);

這實(shí)際上會(huì)聲明下面的對象，它們兩者都可以可選參數(shù)來定制. 在這里，它們兩個(gè)都沒有特定參數(shù)。

1. 一個(gè)類型為direct，且持久化，非自動(dòng)刪除的交換器
2. 采用隨機(jī)生成名稱，且非持久化，私有的，自動(dòng)刪除隊(duì)列

上面的函數(shù)然后使用給定的路由鍵來綁定隊(duì)列和交換器.

注意，當(dāng)只有一個(gè)客戶端時(shí)，這是一種典型聲明隊(duì)列的方式:它不需要一個(gè)已知的名稱，其它的客戶端也不會(huì)使用它(exclusive),并會(huì)被自動(dòng)清除(autodelete).
如果多個(gè)客戶端想共享帶有名稱的隊(duì)列，下面的代碼應(yīng)該更適合:

channel.exchangeDeclare(exchangeName, "direct", true); 
channel.queueDeclare(queueName, true, false, false, null); 
channel.queueBind(queueName, exchangeName, routingKey);

這實(shí)際上會(huì)聲明:

1. 一個(gè)類型為direct，且持久化，非自動(dòng)刪除的交換器
2. 一個(gè)已知名稱，且持久化的，非私有，非自動(dòng)刪除隊(duì)列

注意，Channel API 的方法都是重載的。這些 exchangeDeclare, queueDeclare 和queueBind 都使用的是預(yù)設(shè)行為.
這里也有更多參數(shù)的長形式，它們允許你按需覆蓋默認(rèn)行為，允許你完全控制。

發(fā)由消息

要向交換器中發(fā)布消息,可按下面這樣來使用Channel.basicPublish方法:

byte[] messageBodyBytes = "Hello, world!".getBytes(); 
channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);

為了更好的控制，你可以使用重載方法來指定mandatory標(biāo)志，或使用預(yù)先設(shè)置的消息屬性來發(fā)送消息:

channel.basicPublish(exchangeName, routingKey, mandatory, MessageProperties.PERSISTENT_TEXT_PLAIN,messageBodyBytes);

這會(huì)使用分發(fā)模式2(持久化)來發(fā)送消息, 優(yōu)先級為1，且content-type 為"text/plain".你可以使用Builder類來構(gòu)建你自己的消息屬性對象：

channel.basicPublish(exchangeName, routingKey,new AMQP.BasicProperties.Builder().contentType("text/plain").deliveryMode(2).priority(1).userId("bob").build()),messageBodyBytes);

下面的例子使用自定義的headers來發(fā)布消息:

Map<String, Object> headers = new HashMap<String, Object>(); 
headers.put("latitude",  51.5252949); 
headers.put("longitude", -0.0905493);  
channel.basicPublish(exchangeName, routingKey,new AMQP.BasicProperties.Builder().headers(headers).build()),messageBodyBytes);

下面的例子使用expiration來發(fā)布消息:

channel.basicPublish(exchangeName, routingKey,new AMQP.BasicProperties.Builder().expiration("60000").build()),messageBodyBytes);

BasicProperties is an inner class of the autogenerated holder class AMQP.

Invocations of Channel#basicPublish will eventually block if a resource-driven alarm is in effect.

Channels 和并發(fā)考慮(線程安全性)

Channel 實(shí)例不能在多個(gè)線程間共享。應(yīng)用程序必須在每個(gè)線程中使用不同的channel實(shí)例,而不能將同個(gè)channel實(shí)例在多個(gè)線程間共享。 有些channl上的操作是線程安全的，有些則不是，這會(huì)導(dǎo)致傳輸時(shí)出現(xiàn)錯(cuò)誤的幀交叉。
在多個(gè)線程共享channels也會(huì)干擾Publisher Confirms.

通過訂閱來來接收消息

import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer;

接收消息最高效的方式是用Consumer接口來訂閱。當(dāng)消息到達(dá)時(shí)，它們會(huì)自動(dòng)地進(jìn)行分發(fā)，而不需要顯示地請求。

當(dāng)在調(diào)用Consumers的相關(guān)方法時(shí), 個(gè)別訂閱總是通過它們的consumer tags來確定的, consumer tags可通過客戶端或服務(wù)端來生成，參考 the AMQP specification document.
同一個(gè)channel上的消費(fèi)者必須有不同的consumer tags.

實(shí)現(xiàn)Consumer的最簡單方式是繼承便利類DefaultConsumer.子類可通過在設(shè)置訂閱時(shí)，將其傳遞給basicConsume調(diào)用:

boolean autoAck = false; 
channel.basicConsume(queueName, autoAck, "myConsumerTag",new DefaultConsumer(channel) {          
@Override          
publicvoid handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{              
String routingKey = envelope.getRoutingKey();              
String contentType = properties.getContentType();              
long deliveryTag = envelope.getDeliveryTag();              
// (process the message components here ...)              
channel.basicAck(deliveryTag, false);          
}      
});

在這里，由于我們指定了autoAck = false,因此消費(fèi)者有必要應(yīng)答分發(fā)的消息，最便利的方式是在handleDelivery 方法中處理.

更復(fù)雜的消費(fèi)者可能需要覆蓋更多的方法，實(shí)踐中，handleShutdownSignal會(huì)在channels和connections關(guān)閉時(shí)調(diào)用，handleConsumeOk 會(huì)在其它消費(fèi)者之前

消費(fèi)者可實(shí)現(xiàn)handleCancelOk 和 handleCancel方法來接收顯示和隱式取消操作通知。

你可以使用Channel.basicCancel來顯示地取消某個(gè)特定的消費(fèi)者:

channel.basicCancel(consumerTag);

passing the consumer tag.

消費(fèi)者回調(diào)是在單獨(dú)線程上處理的，這意味著消費(fèi)者可以安全地在Connection或Channel, 如queueDeclare, txCommit, basicCancel或basicPublish上調(diào)用阻塞方法。

每個(gè)Channel都有其自己的dispatch線程.對于一個(gè)消費(fèi)者一個(gè)channel的大部分情況來說，這意味著消費(fèi)者不會(huì)阻擋其它的消費(fèi)者。如果在一個(gè)channel上多個(gè)消費(fèi)者，則必須意識(shí)到長時(shí)間運(yùn)行的消費(fèi)者可能阻擋此channel上其它消費(fèi)者回調(diào)調(diào)度.

獲取單個(gè)消息

要顯示地獲取一個(gè)消息，可使用Channel.basicGet.返回值是一個(gè)GetResponse實(shí)例, 在它之中，header信息(屬性) 和消息body都可以提取:

boolean autoAck = false; 
GetResponse response = channel.basicGet(queueName, autoAck); 
if (response == null) {     
// No message retrieved. 
} else {     
AMQP.BasicProperties props = response.getProps();     
byte[] body = response.getBody();     
long deliveryTag = response.getEnvelope().getDeliveryTag();     ...

因?yàn)?span style="background-color: inherit; color: #333333;">autoAck = false,你必須調(diào)用Channel.basicAck來應(yīng)答你已經(jīng)成功地接收了消息:

channel.basicAck(method.deliveryTag, false); // acknowledge receipt of the message }

處理未路由消息

如果發(fā)布消息時(shí)，設(shè)置了"mandatory"標(biāo)志,但如果消息不能路由的話，broker會(huì)將其返回到發(fā)送客戶端 (通過 AMQP.Basic.Return 命令).

要收到這種返回的通知， clients可實(shí)現(xiàn)ReturnListener接口，并調(diào)用Channel.setReturnListener.如果channel沒有配置return listener,那么返回的消息會(huì)默默地丟棄。

channel.setReturnListener(new ReturnListener() {     
    publicvoid handleBasicReturn(int replyCode,String replyText,String exchange,String routingKey,AMQP.BasicProperties properties,byte[] body)     throws IOException {
         ...     
    } 
});

 return listener將被調(diào)用,例如,如果client使用"mandatory"標(biāo)志向未綁定隊(duì)列的direct類型交換器發(fā)送了消息.

關(guān)閉協(xié)議

import com.rabbitmq.client.ShutdownSignalException; 
import com.rabbitmq.client.ShutdownListener;  
connection.addShutdownListener(new ShutdownListener() {     
public void shutdownCompleted(ShutdownSignalException cause)     {         ...     } }
);

public void shutdownCompleted(ShutdownSignalException cause) {   if (cause.isHardError())   {     
Connection conn = (Connection)cause.getReference();     
if (!cause.isInitiatedByApplication())     {       
Method reason = cause.getReason();       ...     }     ...   } 
else {     Channel ch = (Channel)cause.getReference();     ...   } }

public void brokenMethod(Channel channel) {     if (channel.isOpen())     {         // The following code depends on the channel being in open state.         // However there is a possibility of the change in the channel state         // between isOpen() and basicQos(1) call         ...         channel.basicQos(1);     } }

public void validMethod(Channel channel) {     try {         ...         channel.basicQos(1);     } catch (ShutdownSignalException sse) {         // possibly check if channel was closed         // by the time we started action and reasons for         // closing it         ...     } catch (IOException ioe) {         // check why connection was closed         ...     } }

高級連接選項(xiàng)

ExecutorService es = Executors.newFixedThreadPool(20); Connection conn = factory.newConnection(es); 

Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)                                  , new Address(hostname2, portnumber2)}; Connection conn = factory.newConnection(addrArr); 

import com.google.appengine.api.ThreadManager;  ConnectionFactory cf = new ConnectionFactory(); cf.setThreadFactory(ThreadManager.backgroundThreadFactory()); 

網(wǎng)絡(luò)故障時(shí)自動(dòng)恢復(fù)

ConnectionFactory factory = new ConnectionFactory(); factory.setUsername(userName); factory.setPassword(password); factory.setVirtualHost(virtualHost); factory.setHost(hostName); factory.setPort(portNumber); factory.setAutomaticRecoveryEnabled(true); // connection that will recover automatically Connection conn = factory.newConnection();

ConnectionFactory factory = new ConnectionFactory(); // attempt recovery every 10 seconds factory.setNetworkRecoveryInterval(10000);

ConnectionFactory factory = new ConnectionFactory();  Address[] addresses = {new Address("192.168.1.4"), new Address("192.168.1.5")}; factory.newConnection(addresses);

ConnectionFactory factory = new ConnectionFactory();  Connection conn = factory.newConnection(); factory.setAutomaticRecoveryEnabled(true); factory.setTopologyRecoveryEnabled(false);

未處理異常

關(guān)于connection, channel, recovery, 和consumer lifecycle 的異常將會(huì)委派給exception handler,Exception handler是實(shí)現(xiàn)了ExceptionHandler接口的任何對象. 默認(rèn)情況下,將會(huì)使用DefaultExceptionHandler實(shí)例,它會(huì)將異常細(xì)節(jié)輸出到標(biāo)準(zhǔn)輸出上.

可使用ConnectionFactory#setExceptionHandler來覆蓋原始handler,它將被用于由此factory創(chuàng)建的所有連接:

ConnectionFactory factory = new ConnectionFactory(); cf.setExceptionHandler(customHandler);         

Google App Engine上的RabbitMQ Java Client

在Google App Engine (GAE) 上使用RabbitMQ Java client,必須使用一個(gè)自定義的線程工廠來初始化線程,如使用GAE's ThreadManager. 此外,還需要設(shè)置一個(gè)較小的心跳間隔(4-5 seconds) 來避免InputStream 上讀超時(shí):

ConnectionFactory factory = new ConnectionFactory(); cf.setRequestedHeartbeat(5);         

警告和限制

為了能使拓?fù)浠謴?fù), RabbitMQ Java client 維持了聲明隊(duì)列,交換器,綁定的緩存. 緩存是基于每個(gè)連接的.某些RabbitMQ的特性使得客戶端不能觀察到拓?fù)涞淖兓?如,當(dāng)隊(duì)列因TTL被刪除時(shí). RabbitMQ Java client 會(huì)嘗試在下面的情況中使用緩存實(shí)體失效:

• 當(dāng)隊(duì)列被刪除時(shí).
• 當(dāng)交換器被刪除時(shí).
• 當(dāng)綁定被刪除時(shí).
• 當(dāng)消費(fèi)者在自動(dòng)刪除隊(duì)列上退出時(shí).
• 當(dāng)隊(duì)列或交換器不再綁定到自動(dòng)刪除的交換器上時(shí).

RPC (Request/Reply) 模式

為了便于編程, Java client API提供了一個(gè)使用臨時(shí)回復(fù)隊(duì)列的RpcClient類來提供簡單的RPC-style communication.

此類不會(huì)在RPC參數(shù)和返回值上強(qiáng)加任何特定格式. 它只是簡單地提供一種機(jī)制來向帶特定路由鍵的交換器發(fā)送消息,并在回復(fù)隊(duì)列上等待響應(yīng).

import com.rabbitmq.client.RpcClient;  
RpcClient rpc = new RpcClient(channel, exchangeName, routingKey);

(其實(shí)現(xiàn)細(xì)節(jié)為:請求消息使用basic.correlation_id唯一值字段來發(fā)送消息,并使用basic.reply_to來設(shè)置響應(yīng)隊(duì)列的名稱.)

一旦你創(chuàng)建此類的實(shí)例,你可以使用下面的任意一個(gè)方法來發(fā)送RPC請求:

byte[] primitiveCall(byte[] message); 
String stringCall(String message) Map mapCall(Map message) Map mapCall(Object[] keyValuePairs)

primitiveCall 方法會(huì)將原始byte數(shù)組轉(zhuǎn)換為請求和響應(yīng)的消息體. stringCall只是一個(gè)primitiveCall的簡單包裝,將消息體視為帶有默認(rèn)字符集編碼的String實(shí)例.

mapCall 變種稍為有些復(fù)雜: 它會(huì)將原始java值包含在java.util.Map中,并將其編碼為AMQP 0-9-1二進(jìn)制表示形式,并以同樣的方式來解碼response. (注意:在這里,對一些值對象類型有所限制,具體可參考javadoc.)

所有的編組/解組便利方法都使用primitiveCall來作為傳輸機(jī)制,其它方法只是在它上面的做了一個(gè)封裝.