轮询消费者

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

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

从 2.4.8 版本开始,当使用非零超时时,您可以指定传递给basicConsume方法的参数,该方法用于将消费者与通道关联。例如: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实例的便捷方法,并且实现提供了一种方法来自定义用于创建返回的ObjectMessageConverter:以下列表显示了这些方法

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进行配置。有关更多信息,请参见使用RabbitTemplateMessage转换

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 实现负责处理 receivereply 阶段。在大多数情况下,您只需要提供 ReceiveAndReplyCallback 的实现来执行接收消息的一些业务逻辑,并在需要时构建回复对象或消息。请注意,ReceiveAndReplyCallback 可能会返回 null。在这种情况下,不会发送任何回复,receiveAndReply 的工作方式类似于 receive 方法。这允许同一个队列用于混合消息,其中一些消息可能不需要回复。

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

ReplyToAddressCallback 对于需要自定义逻辑来根据接收到的消息和 ReceiveAndReplyCallback 的回复在运行时确定 replyTo 地址的情况很有用。默认情况下,请求消息中的 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!");
}