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来作为传输机制,其它方法只是在它上面的做了一个封装.