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

• Channel
• Connection
• ConnectionFactory
• Consumer

Connections 和 Channels

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

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

連接到broker

下面的代碼會使用給定的參數連接到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 來設置連接參數:

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

Connection 接口可用來打開一個channel:

Channel channel = conn.createChannel(); 

channel現在可用來發送和接收消息，正如后續章節中描述的一樣.

要斷開連接，只需要簡單地關閉channel和connection:

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

關閉channel被認為是最佳實踐,但在這里不是嚴格必須的 - 當底層連接關閉的時候，channel也會自動關閉.

使用 Exchanges 和 Queues

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

繼續上面的例子,下面的代碼聲明了一個交換器和一個隊列，然后再將它們進行綁定.

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

這實際上會聲明下面的對象，它們兩者都可以可選參數來定制. 在這里，它們兩個都沒有特定參數。

1. 一個類型為direct，且持久化，非自動刪除的交換器
2. 采用隨機生成名稱，且非持久化，私有的，自動刪除隊列

上面的函數然后使用給定的路由鍵來綁定隊列和交換器.

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

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

這實際上會聲明:

1. 一個類型為direct，且持久化，非自動刪除的交換器
2. 一個已知名稱，且持久化的，非私有，非自動刪除隊列

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

發由消息

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

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

為了更好的控制，你可以使用重載方法來指定mandatory標志，或使用預先設置的消息屬性來發送消息:

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

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

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

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

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來發布消息:

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 和并發考慮(線程安全性)

Channel 實例不能在多個線程間共享。應用程序必須在每個線程中使用不同的channel實例,而不能將同個channel實例在多個線程間共享。 有些channl上的操作是線程安全的，有些則不是，這會導致傳輸時出現錯誤的幀交叉。
在多個線程共享channels也會干擾Publisher Confirms.

通過訂閱來來接收消息

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

接收消息最高效的方式是用Consumer接口來訂閱。當消息到達時，它們會自動地進行分發，而不需要顯示地請求。

當在調用Consumers的相關方法時, 個別訂閱總是通過它們的consumer tags來確定的, consumer tags可通過客戶端或服務端來生成，參考 the AMQP specification document.
同一個channel上的消費者必須有不同的consumer tags.

實現Consumer的最簡單方式是繼承便利類DefaultConsumer.子類可通過在設置訂閱時，將其傳遞給basicConsume調用:

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,因此消費者有必要應答分發的消息，最便利的方式是在handleDelivery 方法中處理.

更復雜的消費者可能需要覆蓋更多的方法，實踐中，handleShutdownSignal會在channels和connections關閉時調用，handleConsumeOk 會在其它消費者之前

消費者可實現handleCancelOk 和 handleCancel方法來接收顯示和隱式取消操作通知。

你可以使用Channel.basicCancel來顯示地取消某個特定的消費者:

channel.basicCancel(consumerTag);

passing the consumer tag.

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

每個Channel都有其自己的dispatch線程.對于一個消費者一個channel的大部分情況來說，這意味著消費者不會阻擋其它的消費者。如果在一個channel上多個消費者，則必須意識到長時間運行的消費者可能阻擋此channel上其它消費者回調調度.

獲取單個消息

要顯示地獲取一個消息，可使用Channel.basicGet.返回值是一個GetResponse實例, 在它之中，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();     ...

因為autoAck = false,你必須調用Channel.basicAck來應答你已經成功地接收了消息:

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

處理未路由消息

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

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

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

 return listener將被調用,例如,如果client使用"mandatory"標志向未綁定隊列的direct類型交換器發送了消息.

關閉協議

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         ...     } }

高級連接選項

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()); 

網絡故障時自動恢復

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);

未處理異常

關于connection, channel, recovery, 和consumer lifecycle 的異常將會委派給exception handler,Exception handler是實現了ExceptionHandler接口的任何對象. 默認情況下,將會使用DefaultExceptionHandler實例,它會將異常細節輸出到標準輸出上.

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

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

Google App Engine上的RabbitMQ Java Client

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

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

警告和限制

為了能使拓撲恢復, RabbitMQ Java client 維持了聲明隊列,交換器,綁定的緩存. 緩存是基于每個連接的.某些RabbitMQ的特性使得客戶端不能觀察到拓撲的變化,如,當隊列因TTL被刪除時. RabbitMQ Java client 會嘗試在下面的情況中使用緩存實體失效:

• 當隊列被刪除時.
• 當交換器被刪除時.
• 當綁定被刪除時.
• 當消費者在自動刪除隊列上退出時.
• 當隊列或交換器不再綁定到自動刪除的交換器上時.

RPC (Request/Reply) 模式

為了便于編程, Java client API提供了一個使用臨時回復隊列的RpcClient類來提供簡單的RPC-style communication.

此類不會在RPC參數和返回值上強加任何特定格式. 它只是簡單地提供一種機制來向帶特定路由鍵的交換器發送消息,并在回復隊列上等待響應.

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

(其實現細節為:請求消息使用basic.correlation_id唯一值字段來發送消息,并使用basic.reply_to來設置響應隊列的名稱.)

一旦你創建此類的實例,你可以使用下面的任意一個方法來發送RPC請求:

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

primitiveCall 方法會將原始byte數組轉換為請求和響應的消息體. stringCall只是一個primitiveCall的簡單包裝,將消息體視為帶有默認字符集編碼的String實例.

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

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