跳到主要内容
版本:4.1.1

轮询消费者

Doubao 中英对照 Polling Consumer

AmqpTemplate 本身可用于轮询式 Message 接收。默认情况下,如果没有可用消息,会立即返回 null。没有阻塞。从 1.5 版开始,你可以设置一个以毫秒为单位的 receiveTimeout,接收方法会阻塞至多该时长,等待消息。小于零的值表示无限期阻塞(或至少直到与 broker 的连接断开)。1.6 版引入了 receive 方法的变体,允许在每次调用时传入超时。

警告

由于接收操作会为每条消息创建一个新的 QueueingConsumer,这种技术并不真正适用于高容量环境。对于那些用例,请考虑使用异步消费者或将 receiveTimeout 设为零。

从 2.4.8 版开始,当使用非零超时时,你可以指定传入 basicConsume 方法的参数,该方法用于把消费者与 channel 关联起来。例如:template.addConsumerArg("x-priority", 10)。

有四个简单的 receive 方法可用。与发送侧的 Exchange 一样,有一个方法要求模板本身已直接设置了默认队列属性,还有一个方法在运行时接受一个队列参数。1.6 版引入了接受 timeoutMillis 的变体,可按请求覆盖 receiveTimeout。下面的清单展示了这四个方法的定义:

Message receive() throws AmqpException;

Message receive(String queueName) throws AmqpException;

Message receive(long timeoutMillis) throws AmqpException;

Message receive(String queueName, long timeoutMillis) throws AmqpException;

与发送消息的情况一样,AmqpTemplate 有一些便捷方法用于接收 POJO 而非 Message 实例,并且实现提供了一种方式来定制用于创建所返回 Object 的 MessageConverter。下面的清单展示了这些方法:

Object receiveAndConvert() throws AmqpException;

Object receiveAndConvert(String queueName) throws AmqpException;

Object receiveAndConvert(long timeoutMillis) throws AmqpException;

Object receiveAndConvert(String queueName, long timeoutMillis) throws AmqpException;

从 2.0 版开始,这些方法有一些变体,它们接受一个额外的 ParameterizedTypeReference 参数来转换复杂类型。模板必须配置为使用 SmartMessageConverter。更多信息请参阅用 RabbitTemplate 从消息转换。

与 sendAndReceive 方法类似,从 1.3 版开始,AmqpTemplate 有若干便捷的 receiveAndReply 方法,用于同步地接收、处理消息并回复。下面的清单展示了这些方法定义:

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback)
throws AmqpException;

<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback)
throws AmqpException;

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,
String replyExchange, String replyRoutingKey) throws AmqpException;

<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,
String replyExchange, String replyRoutingKey) throws AmqpException;

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,
ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;

<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,
ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;

AmqpTemplate 的实现负责处理 receive 和 reply 两个阶段。在大多数情况下,你只需提供一个 ReceiveAndReplyCallback 的实现,对接收到的消息执行某些业务逻辑,并在需要时构建回复对象或消息。注意,ReceiveAndReplyCallback 可以返回 null。在这种情况下,不发送回复,receiveAndReply 的行为就像 receive 方法。这让同一个队列可以混合使用多种消息,其中有些消息可能不需要回复。

仅当所提供的回调不是 ReceiveAndReplyMessageCallback(它提供原始消息交换契约)的实例时,才会应用自动的消息(请求和回复)转换。

对于需要根据接收到的消息以及 ReceiveAndReplyCallback 的回复、在运行时用自定义逻辑确定 replyTo 地址的情况,ReplyToAddressCallback 很有用。默认情况下,使用请求消息中的 replyTo 信息来路由回复。

下面的清单展示了一个基于 POJO 的接收与回复示例:

boolean received =
this.template.receiveAndReply(ROUTE, new ReceiveAndReplyCallback<Order, Invoice>() {

public Invoice handle(Order order) {
return processOrder(order);
}
});
if (received) {
log.info("We received an order!");
}