消息接收总是比发送稍微复杂一些。接收消息有两种方法。更简单的方法是使用轮询方法调用一次轮询一条消息。更复杂但更常见的方法是注册一个按需异步接收消息的侦听器。我们将在下两个小节中考虑每种方法的示例。

轮循消费者

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

由于receive操作为每个消息创建一个新的QueueingConsumer，因此这种技术并不适合于大容量环境。考虑为这些用例使用异步使用者或0的receiveTimeout 来适合大容量环境。

有四种简单的接收方法可用。与发送端上的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 ，implementations提供了一种方法来定制用于创建返回对象的MessageConverter:下面的清单显示了这些方法 Object receiveAndConvert() throws AmqpException;

    Object receiveAndConvert(String queueName) throws AmqpException;

    Message receiveAndConvert(long timeoutMillis) throws AmqpException;

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

从2.0版本开始，这些方法的一些变体使用附加的ParameterizedTypeReference参数来转换复杂类型。模板必须配置一个SmartMessageConverter。有关更多信息，请参见从带有RabbitTemplate的消息转换。

与sendAndReceive方法类似，从1.3版本开始，AmqpTemplate有几个方便的receiveandply方法，用于同步接收、处理和回复消息。下面的清单显示了这些方法定义:
 

    <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实现负责接收和应答阶段。在大多数情况下，应该只提供ReceiveAndReplyCallback的实现，以便为接收到的消息执行一些业务逻辑，并在需要时构建一个reply对象或消息。注意，ReceiveAndReplyCallback可能返回null。在本例中，没有send和receiveAndReply，与receive方法类似。这允许对混合消息使用相同的队列，其中一些消息可能不需要回复。

自动消息(请求和应答)转换仅在提供的回调不是

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

}

 

Asynchronous Consumer

Spring AMQP还通过使用@RabbitListener注释支持带注释的侦听器端点，并提供一个开放的基础设施以编程方式注册端点。这是迄今为止设置异步使用者最方便的方法。有关详细信息，请参见注释驱动的侦听器端点。

 

prefetch默认值过去是1，这可能导致有效消费者的利用率不足。从2.0版本开始，默认的预取值现在是250，在大多数常见的场景中，这应该会让消费者保持忙碌，从而提高吞吐量。

然而，在某些情况下，预取值应该很低:

•对于大型消息，特别是处理较慢的消息(消息可能会在客户机进程中累积大量内存)

•当需要严格的消息排序时(在本例中，预取值应该设置为1)

•其他特殊情况

此外，对于低容量消息传递和多个使用者(包括单个侦听器容器实例中的并发性)，您可能希望减少预取，以便在多个使用者之间获得更均匀的消息分布。我们还建议在手动ack模式下使用prefetch = 1。basicAck是一个异步操作，如果在代理上发生了错误(例如，对于相同的交付标记使用双重ack)，您最终会得到批处理中的后续消息，这些消息在代理上没有得到确认，其他消费者可能会看到它们。

参见消息侦听器容器配置。

有关预取的更多背景信息，请参阅本文中的消费者利用率

RabbitMQ和这篇关于排队论的文章。

 

Message Listener

对于异步消息接收，需要一个专用组件(不是AmqpTemplate)。该组件是消息回调使用的容器。我们将在本节后面讨论容器及其属性。不过，首先我们应该看看回调，因为应用程序代码就是在回调中与消息传递系统集成的。有几个回调选项，首先是MessageListener接口的实现，如下清单所示:

public interface MessageListener {
        void onMessage(Message message);

}

如果回调逻辑由于任何原因依赖于AMQP通道实例，则可以使用ChannelAwareMessageListener。它看起来很相似，但是有一个额外的参数。下面的清单显示了ChannelAwareMessageListener接口定义

public interface ChannelAwareMessageListener {
    void onMessage(Message message, Channel channel) throws Exception;

}

In version 2.1, this interface moved from package o.s.amqp.rabbit.core to o.s.amqp.rabbit.listener.api.

 

MessageListenerAdapter

如果希望在应用程序逻辑和消息传递API之间保持更严格的分离，可以依赖于框架提供的适配器实现。这通常被称为“消息驱动的POJO”支持。

版本1.5为POJO消息传递引入了更灵活的机制@RabbitListener注释。有关更多信息，请参见注释驱动的侦听器端点。

在使用适配器时，只需要提供对适配器本身应该调用的实例的引用。下面的例子说明了如何做到这一点:

MessageListenerAdapter listener = new MessageListenerAdapter(somePojo);
    listener.setDefaultListenerMethod("myMethod");

您可以子类化适配器并提供getListenerMethodName()的实现，以便根据消息动态选择不同的方法。该方法有两个参数，originalMessage和extractedMessage，后者是任何转换的结果。默认情况下，配置了SimpleMessageConverter。有关其他可用转换器的更多信息和信息，请参见SimpleMessageConverter。

从1.4.2版本开始，原始消息具有consumerQueue和consumerTag属性，可用于确定接收消息的队列。

从1.5版开始，您可以将使用者队列或标记的映射配置为方法名，以便动态选择要调用的方法。如果映射中没有条目，则返回到默认侦听器方法。默认监听器方法(如果没有设置)是handleMessage。

从2.0版本开始，提供了一个方便的FunctionalInterface。下面的清单显示了FunctionalInterface的定义:

    @FunctionalInterface
    public interface ReplyingMessageListener<T, R> {
        R handleMessage(T t);

        }

 

该接口使用Java 8 lamdas方便适配器的配置，如下例所示:

new MessageListenerAdapter((ReplyingMessageListener<String, String>) data -> {
        ...
        return result;
    }));

 

Container

现在您已经看到了用于消息监听回调的各种选项，我们可以将注意力转向容器。基本上，容器处理“活动”职责，以便侦听器回调可以保持被动。容器是“生命周期”组件的一个例子。它提供了启动和停止方法。在配置容器时，您实际上是在AMQP队列和MessageListener实例之间架起了桥梁。您必须提供对ConnectionFactory和队列名称或队列实例的引用，侦听器应该从中使用消息。

在2.0版本之前，只有一个侦听器容器，SimpleMessageListenerContainer。现在有了第二个容器，DirectMessageListenerContainer。在选择要使用哪个容器时，可能应用的容器和标准之间的差异在选择容器时进行了描述。

下面的清单显示了最基本的例子，它使用SimpleMessageListenerContainer:

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(rabbitConnectionFactory);
    container.setQueueNames("some.queue");
    container.setMessageListener(new MessageListenerAdapter(somePojo));

作为一个“活动”组件，最常见的做法是使用bean定义创建侦听器容器，以便它可以在后台运行。下面的例子展示了一种处理XML的方法:

    <rabbit:listener-container connection-factory="rabbitConnectionFactory">
        <rabbit:listener queues="some.queue" ref="somePojo" method="handle"/>
    </rabbit:listener-container>

The following listing shows another way to do so with XML:

<rabbit:listener-container connection-factory="rabbitConnectionFactory" type=
    "direct">
        <rabbit:listener queues="some.queue" ref="somePojo" method="handle"/>
    </rabbit:listener-container>

前面的两个示例都创建了一个DirectMessageListenerContainer(注意type属性——它默认为simple)。

或者，您可能更喜欢使用Java配置，它看起来类似于前面的代码片段:

    @Configuration
    public class ExampleAmqpConfiguration {

                @Bean

        public SimpleMessageListenerContainer messageListenerContainer() {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(rabbitConnectionFactory());
            container.setQueueName("some.queue");
            container.setMessageListener(exampleListener());
            return container;
        }

              @Bean

        public ConnectionFactory rabbitConnectionFactory() {
            CachingConnectionFactory connectionFactory =new CachingConnectionFactory("localhost");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            return connectionFactory;

}

@Bean

        public MessageListener exampleListener() {
            return new MessageListener() {
                public void onMessage(Message message) {
                    System.out.println("received: " + message);

} };

} }

 

Consumer Priority

从RabbitMQ 3.2版本开始，broker现在支持consumer优先级(请参阅使用RabbitMQ的使用者优先级)。这是通过在使用者上设置 x-priority参数来实现的。SimpleMessageListenerContainer现在支持设置消费者参数，如下面的例子所示:

container.setConsumerArguments(Collections.
    <String, Object> singletonMap("x-priority", Integer.valueOf(10)));

为了方便起见，命名空间在listener元素上提供了priority属性，如下例所示:

    <rabbit:listener-container connection-factory="rabbitConnectionFactory">
        <rabbit:listener queues="some.queue" ref="somePojo" method="handle" priority=
    "10" />
    </rabbit:listener-container>

从版本1.3开始，您可以修改容器在运行时侦听的队列。参见侦听器容器队列。

 

auto-delete Queues

当容器被配置为侦听自动删除队列时，队列有一个x-expires选项，或者代理上配置了生存时间策略，当容器停止时(即最后一个使用者被取消时)，代理将删除队列。在版本1.3之前，由于队列丢失，容器无法重新启动。RabbitAdmin只在连接关闭或打开时自动重新声明队列，而在容器停止和启动时不会这样做。

从1.3版本开始，容器使用RabbitAdmin在启动期间重新声明任何丢失的队列。

您还可以使用条件声明(参见条件声明)和auto-startup="false" admin来延迟队列声明，直到容器启动。下面的例子说明了如何做到这一点:

    <rabbit:queue id="otherAnon" declared-by="containerAdmin" />

    <rabbit:direct-exchange name="otherExchange" auto-delete="true" declared-by=
    "containerAdmin">

        <rabbit:bindings>
            <rabbit:binding queue="otherAnon" key="otherAnon" />

        </rabbit:bindings>
    </rabbit:direct-exchange>

    <rabbit:listener-container id="container2" auto-startup="false">
        <rabbit:listener id="listener2" ref="foo" queues="otherAnon" admin=
    "containerAdmin" />
    </rabbit:listener-container>

    <rabbit:admin id="containerAdmin" connection-factory="rabbitConnectionFactory"
        auto-startup="false" />

在本例中，队列和交换由containmin声明，它具有auto- startup="false"，因此在上下文初始化期间不会声明元素。同样，容器也不是出于相同的原因启动的。稍后启动容器时，它使用对contain根除min的引用来声明元素。

 

Batched Messages

批处理消息由侦听器容器自动反批处理(使用springBatchFormat消息头)。拒绝批处理中的任何消息都会导致整个批处理被拒绝。有关批处理的更多信息，请参见批处理。

 

Consumer Events

每当侦听器(使用者)遇到某种故障时，容器就会发布应用程序事件。事件ListenerContainerConsumerFailedEvent具有以下属性:

容器:消费者遇到问题的侦听器容器。

•原因:失败的文本原因。

•致命:指示失败是否致命的布尔值。对于非致命异常，容器尝试根据recoveryInterval或recoveryBackoff(对于SimpleMessageListenerContainer)或monitorInterval(对于DirectMessageListenerContainer)重新启动使用者。

•可扔的:被抓住的可扔的东西。

可以通过实现使用这些事件

ApplicationListener < ListenerContainerConsumerFailedEvent >。

当concurrentConsumers大于1时，所有使用者都会发布系统范围的事件(例如连接失败)。

如果使用者失败，因为它的队列在缺省情况下是专用的，并且发布事件，那么将发出警告日志。要更改此日志记录行为，请在SimpleMessageListenerContainer实例的exclusive veconsumerexceptionlogger属性中提供一个定制的ConditionalExceptionLogger。请参见记录通道关闭事件。

致命错误总是记录在错误级别。这是不可修改的。在容器生命周期的不同阶段还发布了其他几个事件:

AsyncConsumerStartedEvent:当消费者启动时。

AsyncConsumerRestartedEvent:当消费者失败后重新启动-

SimpleMessageListenerContainer。

AsyncConsumerTerminatedEvent:当用户正常停止时。

AsyncConsumerStoppedEvent:当消费者停止时——SimpleMessageListenerContainer only。

•ConsumeOkEvent:当从代理接收到consumeOk时，包含队列名称和consumerTag

•ListenerContainerIdleEvent:参见检测空闲异步消费者

 

 

Consumer Tags

您可以提供一种策略来生成消费者标记。默认情况下，consumer标记broker 生成。下面的清单显示了ConsumerTagStrategy接口定义

    public interface ConsumerTagStrategy {
        String createConsumerTag(String queue);

}

The queue is made available so that it can (optionally) be used in the tag. See Message Listener Container Configuration.

 

Annotation-driven Listener Endpoints

异步接收消息的最简单方法是使用带注释的侦听器端点基础设施。简而言之，它允许您将托管bean的方法公开为rabbitmq侦听器端点。下面的例子展示了如何使用@RabbitListener注释:

@Component

    public class MyService {

        @RabbitListener(queues = "myQueue")
        public void processOrder(String data) {

... }

}

上一个示例的思想是，每当消息在名为myQueue的队列上可用时，都会相应地调用processOrder方法(在本例中，使用消息的有效负载)。

带注释的端点基础设施在后台为每个带注释的方法通过使用RabbitListenerContainerFactory创建一个消息侦听器容器。

在前面的示例中，myQueue必须已经存在，并被绑定到某个exchange。只要应用程序上下文中存在一个RabbitAdmin，就可以自动声明和绑定队列。

可以为注释属性(队列等)指定属性占位符(${some.property})或SpEL表达式(#{someExpression})。有关为什么可能使用SpEL而不是属性占位符的示例，请参阅侦听多个队列。下面的清单显示了三个如何声明rabbit侦听器的示例:

@Component

    public class MyService {

      @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "myQueue", durable = "true"),
            exchange = @Exchange(value = "auto.exch", ignoreDeclarationExceptions =
    "true"),
            key = "orderRoutingKey")
      )
      public void processOrder(Order order) {

... }

      @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            exchange = @Exchange(value = "auto.exch"),
            key = "invoiceRoutingKey")
      )

      public void processInvoice(Invoice invoice) {
        ...

}

      @RabbitListener(queuesToDeclare = @Queue(name = "${my.queue}", durable = "true"
    ))
      public String handleWithSimpleDeclare(String data) {
          ...

} }

在第一个示例中，如果需要，队列myQueue将与交换一起自动(持久)声明，并使用路由键绑定到交换。在第二个示例中，声明并绑定了一个匿名(排他的、自动删除的)队列。可以提供多个QueueBinding条目，让侦听器侦听多个队列。在第三个示例中，从属性my检索到一个具有名称的队列。如果需要，将使用缺省绑定声明queue，并使用队列名称作为路由键将其绑定到缺省交换器。

自2.0版以来，@Exchange注释支持任何交换类型，包括自定义。有关更多信息，请参见AMQP概念。

当您需要更高级的配置时，可以使用普通的@Bean定义。

注意第一个例子中的ignoredeclaration异常。例如，这允许绑定到可能具有不同设置(例如内部设置)的现有交换器。默认情况下，现有交换器的属性必须匹配。

从2.0版开始，您现在可以使用多个路由键将队列绑定到exchange，如下面的示例所示:

key = { "red", "yellow" }

您还可以在@QueueBinding注释中为队列、交换和绑定指定参数，如下面的示例所示:

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "auto.headers", autoDelete = "true",

                            arguments = @Argument(name = "x-message-ttl", value =

"10000",

                                                    type = "java.lang.Integer")),
            exchange = @Exchange(value = "auto.headers", type = ExchangeTypes.HEADERS,

    autoDelete = "true"),
            arguments = {

                    @Argument(name = "x-match", value = "all"),
                    @Argument(name = "thing1", value = "somevalue"),
                    @Argument(name = "thing2")

}) )

    public String handleWithHeadersExchange(String foo) {
        ...

}

 

注意，x-message-ttl参数为队列设置为10秒。由于参数类型不是字符串，我们必须指定它的类型——在本例中是Integer。与所有此类声明一样，如果队列已经存在，则参数必须与队列上的参数匹配。对于报头交换，我们设置绑定参数来匹配那些将thing1报头设置为somevalue的消息，并且thing2报头必须存在。x-match参数意味着必须满足这两个条件。

参数名、值和类型可以是属性占位符(${…})或SpEL表达式(#{…})。名称必须解析为字符串。type的表达式必须解析为类或类的完全限定名。该值必须解析为可以由DefaultConversionService转换为类型的值(如前面示例中的x-message-ttl)。

如果名称解析为null或空字符串，则忽略@参数。

 

Meta-annotations

有时，您可能希望对多个侦听器使用相同的配置。要减少样板配置，可以使用元注释创建自己的侦听器注释。下面的例子说明了如何做到这一点:

    @Target({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE})
    @Retention(RetentionPolicy.RUNTIME)
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,
            exchange = @Exchange(value = "metaFanout", type = ExchangeTypes.FANOUT)))
    public @interface MyAnonFanoutListener {
    }

    public class MetaListener {

        @MyAnonFanoutListener
        public void handle1(String foo) {
            ...

}

        @MyAnonFanoutListener
        public void handle2(String foo) {
            ...

} }

在前面的示例中，@ MyAnonFanoutListener注释创建的每个侦听器都将一个匿名的自动删除队列绑定到fanout exchange metaFanout。元注释机制很简单，因为不检查用户定义的注释上的属性——因此不能覆盖来自元注释的设置。当您需要更高级的配置时，可以使用普通的@Bean定义。

 

Enable Listener Endpoint Annotations

要启用对@RabbitListener注释的支持，可以将@EnableRabbit添加到@Configuration类之一。下面的例子说明了如何做到这一点:

    @Configuration
    @EnableRabbit
    public class AppConfig {

@Bean

        public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory());
            factory.setConcurrentConsumers(3);
            factory.setMaxConcurrentConsumers(10);
            return factory;
        }

}

从2.0版本开始，也可以使用DirectMessageListenerContainerFactory。它创建DirectMessageListenerContainer实例。

有关帮助您在SimpleRabbitListenerContainerFactory和DirectRabbitListenerContainerFactory之间进行选择的信息，请参见选择容器。

默认情况下，基础设施寻找一个名为rabbitListenerContainerFactory的bean作为工厂用于创建消息侦听器容器的源。在本例中，忽略RabbitMQ基础设施设置，processOrder方法可以使用三个线程的核心轮询大小和十个线程的最大池大小来调用。

您可以自定义侦听器容器工厂来为每个注释使用，也可以通过实现RabbitListenerConfigurer接口来配置显式默认值。只有在没有特定容器工厂注册了至少一个端点时，才需要默认值。有关详细信息和示例，请参阅Javadoc。

容器工厂提供了添加MessagePostProcessor实例的方法，这些实例在接收消息(调用侦听器之前)和发送响应之前应用。

从2.0.6版本开始，您可以向侦听器容器工厂添加RetryTemplate和RecoveryCallback。它用于发送回复。当重试耗尽时，将调用RecoveryCallback。您可以使用SendRetryContextAccessor从上下文获取信息。下面的例子说明了如何做到这一点:

 

    factory.setRetryTemplate(retryTemplate);
    factory.setReplyRecoveryCallback(ctx -> {

        Message failed = SendRetryContextAccessor.getMessage(ctx);
        Address replyTo = SendRetryContextAccessor.getAddress(ctx);
        Throwable t = ctx.getLastThrowable();
        ...

        return null;
    });

如果喜欢XML配置，可以使用元素。任何带有@RabbitListener注释的bean会被检测到。

对于SimpleRabbitListenerContainer实例，可以使用类似于下面的XML:

<rabbit:annotation-driven/>

    <bean id="rabbitListenerContainerFactory"
          class=
    "org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="concurrentConsumers" value="3"/>
        <property name="maxConcurrentConsumers" value="10"/>

         </bean>

对于DirectMessageListenerContainer实例，可以使用类似于下面的XML:

<rabbit:annotation-driven/>

    <bean id="rabbitListenerContainerFactory"
          class=

    "org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="consumersPerQueue" value="3"/>

</bean>

 

从2.0版本开始，@RabbitListener注释有一个并发属性。它支持SpEL表达式(#{…})和属性占位符(${…})。它的含义和允许的值取决于容器类型，如下:

•对于DirectMessageListenerContainer，该值必须是一个整数值，该整数值设置容器上的consumersPerQueue属性。

•对于SimpleRabbitListenerContainer，该值可以是一个整数值，它设置容器上的concurrentConsumers属性，或者它可以有m-n形式，其中m是concurrentConsumers属性，n是maxConcurrentConsumers属性。

在这两种情况下，此设置都会覆盖工厂上的设置。以前，如果有需要不同并发性的侦听器，就必须定义不同的容器工厂。

 

 

Message Conversion for Annotated Methods

在调用侦听器之前，管道中有两个转换步骤。第一步使用MessageConverter将传入的Spring AMQP消息转换为Spring message消息。当调用目标方法时，如果需要，消息有效负载将转换为方法参数类型。

第一步的默认MessageConverter是一个Spring AMQP SimpleMessageConverter，它处理到字符串和java.io的转换。可序列化的对象。所有其他的都保留为byte[]。在下面的讨论中，我们将其称为“消息转换器”。

第二步的默认转换器是GenericMessageConverter，它将委托给转换服务(DefaultFormattingConversionService的实例)。在下面的讨论中，我们将其称为“方法参数转换器”。

要更改消息转换器，可以将其作为属性添加到容器工厂bean。下面的例子说明了如何做到这一点:

@Bean

    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new

    SimpleRabbitListenerContainerFactory();
        ...

        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        ...
        return factory;

}

这配置了一个Jackson2转换器，它期望头信息能够指导转换。

您还可以使用ContentTypeDelegatingMessageConverter，它可以处理不同内容类型的转换。

在大多数情况下，没有必要自定义方法参数转换器，除非您想使用自定义的ConversionService。

在1.6之前的版本中，必须在消息头中提供转换JSON的类型信息，或者需要自定义类映射器。从version 1.6开始，如果没有类型信息头，则可以从目标方法参数推断类型。

这种类型推断只适用于方法级别的@RabbitListener。

If you wish to customize the method argument converter, you can do so as follows:

@Configuration
    @EnableRabbit
    public class AppConfig implements RabbitListenerConfigurer {

...

@Bean

        public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
            DefaultMessageHandlerMethodFactory factory = new

    DefaultMessageHandlerMethodFactory();
            factory.setMessageConverter(new GenericMessageConverter

    (myConversionService()));
            return factory;

}

@Bean

        public ConversionService myConversionService() {
            DefaultConversionService conv = new DefaultConversionService();
            conv.addConverter(mySpecialConverter());
            return conv;

}

@Override

        public void configureRabbitListeners(RabbitListenerEndpointRegistrar
    registrar) {

            registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
        }

... }

 

Programmatic Endpoint Registration

RabbitListenerEndpoint提供了一个rabbit 端点的模型，并负责为该模型配置容器。除了RabbitListener注释检测到的端点之外，该基础设施还允许您以编程方式配置端点。下面的例子说明了如何做到这一点:

@Configuration
@EnableRabbit
public class AppConfig implements RabbitListenerConfigurer {

@Override

    public void configureRabbitListeners(RabbitListenerEndpointRegistrar
registrar) {

            SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint(

            endpoint.setQueueNames("anotherQueue");
            endpoint.setMessageListener(message -> {

                // processing

});

            registrar.registerEndpoint(endpoint);
        }

}

在前面的示例中，我们使用了SimpleRabbitListenerEndpoint，它提供了要调用的实际MessageListener，但是您也可以构建自己的端点变体来描述自定义调用机制。

应该注意的是，您完全可以跳过@RabbitListener的使用，而通过RabbitListenerConfigure以编程方式注册端点

 

Annotated Endpoint Method Signature

到目前为止，我们已经在端点中注入了一个简单的字符串，但是它实际上可以有一个非常灵活的方法签名。follwoing例子重写了它，用自定义头注入订单:

@Component

    public class MyService {

        @RabbitListener(queues = "myQueue")

        public void processOrder(Order order, @Header("order_type") String orderType)
    {

... }

}

 

下面的列表显示了可以注入侦听器端点的主要元素:

•原始的org.springframework.amqp.core.Message 。

• com.rabbitmq.client.Channel  接收消息的通道。

the org.springframework.messaging.Message representing the incoming AMQP message. Note that this message holds both the custom and the standard headers (as defined by AmqpHeaders).

 

@ header注释的方法参数来提取特定的标头值，包括标准的AMQP标头。

•@ headers -带注释的参数，也必须可分配给java.util.map 用于访问所有标题的映射。

 

非注释元素(即消息和通道)被认为是有效负载。您可以通过使用@Payload注释参数来显式地实现这一点。您还可以通过添加额外的@Valid来打开验证。

注入Spring的消息抽象的能力对于从存储在特定于传输的消息中的所有信息中获益尤其有用，而无需依赖于特定于传输的API。下面的例子说明了如何做到这一点:

    @RabbitListener(queues = "myQueue")
    public void processOrder(Message<Order> order) { ...
    }

方法参数的处理由DefaultMessageHandlerMethodFactory提供，您可以进一步定制它来支持其他方法参数。转换和验证支持也可以在那里定制。

例如，如果我们想在处理订单之前确保订单是有效的，我们可以用@Valid标注有效负载，并配置必要的验证器，如下所示:

    @Configuration
    @EnableRabbit
    public class AppConfig implements RabbitListenerConfigurer {

@Override

        public void configureRabbitListeners(RabbitListenerEndpointRegistrar
    registrar) {

            registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
        }

@Bean

        public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
            DefaultMessageHandlerMethodFactory factory = new

    DefaultMessageHandlerMethodFactory();
            factory.setValidator(myValidator());
            return factory;

} }

 

Listening to Multiple Queues

使用queues属性时，可以指定关联容器可以侦听多个队列。您可以使用@Header注释使接收消息的队列名称对POJO方法可用。下面的例子说明了如何做到这一点:

@Component

    public class MyService {

        @RabbitListener(queues = { "queue1", "queue2" } )

        public void processOrder(String data, @Header(AmqpHeaders.CONSUMER_QUEUE)
    String queue) {

... }

}

 

从1.5版本开始，您可以使用属性占位符和SpEL将队列名称外部化。下面的例子说明了如何做到这一点:

@Component

    public class MyService {

        @RabbitListener(queues =
    "#{'${property.with.comma.delimited.queue.names}'.split(',')}" )

        public void processOrder(String data, @Header(AmqpHeaders.CONSUMER_QUEUE)
    String queue) {

... }

}

 

在1.5版本之前，只能以这种方式指定一个队列。每个队列都需要一个单独的属性。

 

Reply Management

MessageListenerAdapter中的现有支持已经允许您的方法具有非空返回类型。在这种情况下，调用的结果被封装在消息中，发送到原始消息的ReplyToAddress头中指定的地址中，或者在侦听器上配置的默认地址中。您可以使用消息传递抽象的@SendTo注释设置默认地址。

假设我们的processOrder方法现在应该返回一个OrderStatus，我们可以这样写来自动发送一个回复:

    @RabbitListener(destination = "myQueue")
    @SendTo("status")
    public OrderStatus processOrder(Order order) {

        // order processing

        return status;
    }

如果需要以与传输无关的方式设置附加头，可以返回一条消息，如下所示:

    @RabbitListener(destination = "myQueue")
    @SendTo("status")
    public Message<OrderStatus> processOrder(Order order) {

        // order processing

        return MessageBuilder
            .withPayload(status)
            .setHeader("code", 1234)
            .build();

}

 

@SendTo值假设为响应交换和路由键对，它们遵循exchange/routingKey模式，其中可以省略其中的一个部分。有效值如下:

thing1/thing2:回复和路由键。thing1/: replyTo exchange和默认(空)routingKey。thing2或/thing2: replyTo routingKey和默认(空)交换。/或空:replyTo默认交换和默认路由键。

此外，可以使用@SendTo而不使用value属性。这种情况等于一个空的sendTo模式。@SendTo仅在入站消息没有replyToAddress属性时才使用。

从1.5版本开始，@SendTo值可以是一个bean初始化SpEL表达式，如下面的示例所示:

@RabbitListener(queues = "test.sendTo.spel")
    @SendTo("#{spelReplyTo}")
    public String capitalizeWithSendToSpel(String foo) {

        return foo.toUpperCase();
    }

...

@Bean

    public String spelReplyTo() {
        return "test.sendTo.reply.spel";

}

表达式必须计算为字符串，字符串可以是简单的队列名称(发送到默认交换器)，也可以是前面示例中讨论的表单exchange/routingKey。

备用运行时SpEL表达式(在下一个示例之后描述)。

从1.6版开始，@SendTo可以是一个SpEL表达式，在运行时根据请求和响应进行计算，如下面的示例所示:

    @RabbitListener(queues = "test.sendTo.spel")
    @SendTo("!{'some.reply.queue.with.' + result.queueName}")
    public Bar capitalizeWithSendToSpel(Foo foo) {

        return processTheFooAndReturnABar(foo);
    }

SpEL表达式的运行时属性用!{…}分隔符。表达式的计算上下文#root对象有三个属性:

•请求:操作系统amqm .core。消息请求对象。

•消息来源:os . message . message >后转换。•结果:方法结果。

上下文有一个map属性访问器、一个标准类型转换器和一个bean解析器，它允许引用上下文中的其他bean(例如，@ somebeanname .行列式plyq (request, result))。

总之,# {…}在初始化期间计算一次，#root对象作为应用程序上下文。bean由它们的名称引用。! {…}在运行时为每个消息求值，根对象具有前面列出的属性。bean用它们的名称引用，前面加上@。

从2.1版本开始，还支持简单的属性占位符(例如${some.reply.to})。对于较早的版本，可以使用以下代码作为参考，如下例所示:

@RabbitListener(queues = "foo")
    @SendTo("#{environment['my.send.to']}")
    public String listen(Message in) {

...

return ... }

 

Multi-method Listeners

从1.5.0版本开始，您可以在类级别指定@RabbitListener注释。加上新的@RabbitHandler注释，这允许单个侦听器根据传入消息的有效负载类型调用不同的方法。这最好用一个例子来描述:

    @RabbitListener(id="multi", queues = "someQueue")
    @SendTo("my.reply.queue")
    public class MultiListenerBean {

        @RabbitHandler

        public String thing2(Thing2 thing2) {
            ...

}

        @RabbitHandler

        public String cat(Cat cat) {
            ...

}

        @RabbitHandler

        public String hat(@Header("amqp_receivedRoutingKey") String rk, @Payload Hat
    hat) {

... }

        @RabbitHandler(isDefault = true)
        public String defaultMethod(Object object) {

... }

}

 

在本例中，如果转换的有效负载是Thing2、Cat或Hat，则调用单个@RabbitHandler方法。您应该了解，系统必须能够根据负载类型识别惟一的方法。检查类型是否可分配给没有注释或使用@Payload注释进行注释的单个参数。注意，同样的方法签名也适用，正如前面描述的@RabbitListener方法级别中所讨论的那样。

从2.0.3版本开始，可以将@RabbitHandler方法指定为默认方法，如果其他方法上没有匹配，则调用该方法。最多只能指定一种方法。

@RabbitHandler仅用于处理转换后的消息有效负载，如果希望接收未转换的原始消息对象，必须在方法上使用@RabbitListener，而不是类。

 

 

@Repeatable @RabbitListener

从1.6版开始，@RabbitListener注释被标记为@Repeatable。这意味着注释可以多次出现在同一个带注释的元素(方法或类)上。在本例中，为每个注释创建一个单独的侦听器容器，每个注释都调用相同的侦听器@Bean。可重复注释可以与Java 8或更高版本一起使用。在使用Java 7或更早版本时，可以通过使用@RabbitListener“容器”注释(带有@RabbitListener注释数组)实现相同的效果。

 

 

Proxy @RabbitListener and Generics

 

如果您的服务打算代理(例如@Transactional)，那么当接口具有通用参数时，您应该记住一些注意事项。考虑下面的例子

interface TxService<P> {
       String handle(P payload, String header);

    }
    static class TxServiceImpl implements TxService<Foo> {

        @Override
        @RabbitListener(...)
        public String handle(Foo foo, String rk) {

... }

}

 

对于通用接口和特定的实现，您必须切换到CGLIB目标类代理，因为接口句柄方法的实际实现是桥接方法。在事务管理的情况下，CGLIB的使用是通过使用一个注释选项来配置的:@EnableTransactionManagement(proxyTargetClass = true)。在这种情况下，所有的注解都必须在实现的目标方法上声明，如下例所示:

static class TxServiceImpl implements TxService<Foo> {

        @Override
        @Transactional
        @RabbitListener(...)
        public String handle(@Payload Foo foo, @Header("amqp_receivedRoutingKey")

    String rk) {
            ...

} }

 

Handling Exceptions

 

默认情况下，如果带注释的侦听器方法抛出异常，则会将异常抛出给容器，然后根据容器和代理配置对消息进行重新请求、重新交付、丢弃或路由到死信交换。没有任何内容返回给发送方。

从2.0版本开始，@RabbitListener注释有两个新属性:errorHandler和returnexception。

默认情况下不配置这些。

您可以使用errorHandler来提供RabbitListenerErrorHandler实现的bean名。这个功能接口有一个方法，如下:

@FunctionalInterface

  public interface RabbitListenerErrorHandler {
      Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> . message,ListenerExecutionFailedException exception) throws Exception;

}

 

如您所见，您可以访问从容器接收到的原始消息以及由消息转换器生成的对象spring message，以及侦听器抛出的异常(包装在ListenerExecutionFailedException中)。错误处理程序可以返回一些结果(作为响应发送)，也可以抛出原始异常或新异常(根据returnexception设置，将其抛出到容器或返回给发送方)。

returnExceptions属性为真时，会将异常返回给发送方。异常被包装在RemoteInvocationResult对象中。在发送方端，有一个可用的RemoteInvocationAwareMessageConverterAdapter，如果将其配置到RabbitTemplate中，它将重新抛出服务器端异常，并将其封装在AmqpRemoteException中。通过合并服务器和客户端堆栈跟踪，可以合成服务器异常的堆栈跟踪。

 

这种机制通常只适用于默认的SimpleMessageConverter，它使用Java序列化。异常通常不是“JSON友好的”，并且不能序列化为JSON。如果您使用JSON，请考虑在抛出异常时使用errorHandler返回其他一些对JSON友好的错误对象。

In version 2.1, this interface moved from package o.s.amqp.rabbit.listener to o.s.amqp.rabbit.listener.api.

 

从2.1.7版本开始，通道在消息消息头中可用;这允许您在使用 AcknowledgeMode.MANUAL对失败的消息进行ack或nack。

public Object handleError(Message amqpMessage, org.springframework.messaging
.Message<?> message,
          ListenerExecutionFailedException exception) {
              ...
    Long.class),
 }

 

Container Management

为注释创建的容器不会在应用程序上下文中注册。您可以通过调用RabbitListenerEndpointRegistry bean上的getListenerContainers()来获得所有容器的集合。然后，您可以遍历这个集合，例如，停止或启动所有容器，或者调用注册表本身上的生命周期方法，这些方法将调用每个容器上的操作。

您还可以通过使用单个容器的id(使用getListenerContainer(字符串id))获得对该容器的引用，例如，上面代码片段创建的容器的register .getListenerContainer(“multi”)。

从1.5.2版本开始，您可以使用getListenerContainerIds()获取已注册容器的id值。

从1.5版开始，现在可以在RabbitListener端点上为容器分配一个组。这提供了一种机制来获取对容器子集的引用。添加组属性会导致集合类型的bean在具有组名称的上下文中注册。

Using Container Factories

引入侦听器容器工厂是为了支持@RabbitListener并向RabbitListenerEndpointRegistry注册容器，正如编程端点注册中讨论的那样。

从2.1版开始，它们可以用来创建任何侦听器容器——甚至是没有侦听器的容器(如Spring Integration中使用的容器)。当然，必须在启动容器之前添加侦听器。

有两种方法可以创建这样的容器:

•创建后添加侦听器

下面的示例展示了如何使用SimpleRabbitListenerEndpoint创建侦听器容器:

@Bean

    public SimpleMessageListenerContainer factoryCreatedContainerSimpleListener(
            SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory) {

        SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint();
        endpoint.setQueueNames("queue.1");
        endpoint.setMessageListener(message -> {

... });

        return rabbitListenerContainerFactory.createListenerContainer(endpoint);
    }

 

The following example shows how to add the listener after creation:

 

@Bean

    public SimpleMessageListenerContainer factoryCreatedContainerNoListener(
            SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory) {

        SimpleMessageListenerContainer container = rabbitListenerContainerFactory
    .createListenerContainer();

        container.setMessageListener(message -> {
            ...

        });
        container.setQueueNames("test.no.listener.yet");
        return container;

}

在这两种情况下，侦听器都可以是ChannelAwareMessageListener，因为它现在是MessageListener的子接口。

如果您希望创建几个具有类似属性的容器，或者使用预配置的容器工厂(如Spring Boot auto configuration提供的容器工厂)，或者两者都使用，那么这些技术是非常有用的。

以这种方式创建的容器是普通的@Bean实例，不会在RabbitListenerEndpointRegistry中注册。

 

Asynchronous @RabbitListener Return Types

从2.1版开始，@RabbitListener (and @RabbitHandler)  方法 可以指定 异步响应类型 ListenableFuture<?> and Mono<?>, 让响应异步发送。

侦听器容器工厂必须配置为 AcknowledgeMode.MANUAL。使consumer线程不会自动确认消息;相反，在异步操作完成时对消息进行ack或nack。当异步结果完成并带有错误时，是否重新请求消息取决于抛出的异常类型、容器配置和容器错误处理程序。默认情况下，消息将被重新请求，除非容器的defaultRequeueRejected属性设置为false。如果完成异步结果但是跑了一场AmqpRejectAndDontRequeueException，则不会对消息进行重新请求。如果侦听器方法中发生异常，阻止创建异步结果对象，则必须捕获该异常并返回适当的返回对象，该对象将导致消息被确认或重新请求。

 

Threading and Asynchronous Consumers

异步消费者涉及到了许多不同的线程。

在SimpleMessageListenerContainer中配置的TaskExecutor中的线程用于在RabbitMQ客户机交付新消息时调用MessageListener。如果没有配置，则使用SimpleAsyncTaskExecutor。如果使用池执行器，则需要确保池大小足以处理配置的并发性。使用DirectMessageListenerContainer, MessageListener将直接在RabbitMQ客户机线程上调用。在本例中，taskExecutor用于监视consumer的任务。

当使用默认SimpleAsyncTaskExecutor时，对于调用侦听器的线程，在threadNamePrefix中使用侦听器容器beanName。这对日志分析很有用。我们通常建议始终在日志追加器配置中包含线程名。当通过容器上的TaskExecutor属性特别提供TaskExecutor时，就按原样使用它，无需修改。建议使用类似的技术来命名由自定义TaskExecutor bean定义创建的线程，以帮助在日志消息中进行线程标识。

在CachingConnectionFactory中配置的执行器在创建连接时被传递到RabbitMQ客户机，其线程用于向侦听器容器传递新消息。如果没有配置，客户机将使用池大小为5的内部线程池执行器。

使用DirectMessageListenerContainer，您需要确保连接工厂配置了一个任务执行器，该执行器具有足够的线程来支持使用该工厂的所有侦听器容器之间所需的并发性。默认池大小只有5。

RabbitMQ客户机使用ThreadFactory为低级I/O(套接字)操作创建线程。要修改这个工厂，您需要配置底层RabbitMQ ConnectionFactory，正如在配置底层客户机连接工厂中讨论的那样。

 

Choosing a Container

版本2.0引入了DirectMessageListenerContainer (DMLC)。以前，只有SimpleMessageListenerContainer (SMLC)可用。SMLC为每个使用者使用一个内部队列和一个专用线程。如果将容器配置为侦听多个队列，则使用同一个使用者线程处理所有队列。并发性由concurrentconsumer和其他属性控制。当消息从RabbitMQ客户机到达时，客户机线程通过队列将消息传递给消费者线程。之所以需要这种体系结构，是因为在RabbitMQ客户机的早期版本中，不可能有多个并发交付。新版本的客户机具有经过修改的线程模型，现在可以支持并发。这允许引入DMLC，其中侦听器现在直接在RabbitMQ客户机线程上调用。因此，它的架构实际上比SMLC“更简单”。但是，这种方法有一些限制，并且SMLC的某些特性在DMLC中不可用。此外，并发性由consumersPerQueue(和客户机库的线程池)控制。concurrentConsumers和关联的属性在此容器中不可用。

以下功能是可用的SMLC，但不是DMLC:

•txSize:使用SMLC，您可以将其设置为控制在一个事务中交付了多少消息，或者减少ack的数量，但是这可能会导致失败后重复交付的数量增加。(DMLC确实有messagesPerAck，您可以使用它来减少ack，与txSize和SMLC相同，但是它不能用于事务——每个消息都在单独的事务中交付和打包)。

•maxConcurrentConsumers和consumer scale interval或触发器——DMLC中没有自动缩放功能。但是，它允许您以编程方式更改consumersPerQueue属性，并相应地调整使用者。

然而，与SMLC相比，DMLC有以下好处:

•在运行时添加和删除队列更有效。使用SMLC，将重新启动整个使用者线程(所有使用者都被取消并重新创建)。使用DMLC，未受影响的消费者不会被取消。

•避免了RabbitMQ客户端线程和消费者线程之间的上下文切换。

•线程是跨消费者共享的，而不是在SMLC中为每个消费者都有一个专用的线程。但是，请参阅关于线程和异步使用者中的连接工厂配置的重要说明。

有关应用于每个容器的配置属性的信息，请参阅消息侦听器容器配置。

 

Detecting Idle Asynchronous Consumers

虽然效率很高，但是异步使用者的一个问题是检测何时空闲—如果在一段时间内没有消息到达，用户可能希望采取一些行动。

从1.6版开始，现在可以配置侦听器容器，以便在没有消息传递的情况下发布ListenerContainerIdleEvent。当容器处于空闲状态时，每隔几毫秒就会发布一个事件。

73

要配置此功能，请在容器上设置idleEventInterval。下面的例子展示了如何在XML和Java中实现这一点(对于SimpleMessageListenerContainer和SimpleRabbitListenerContainerFactory):

<rabbit:listener-container connection-factory="connectionFactory"
            ...

            idle-event-interval="60000"
            ...
            >

        <rabbit:listener id="container1" queue-names="foo" ref="myListener" method=
    "handle" />

    </rabbit:listener-container>

 

@Bean

    public SimpleMessageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer

    (connectionFactory);
        ...

        container.setIdleEventInterval(60000L);
        ...
        return container;

}

@Bean

    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new

    SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(rabbitConnectionFactory());
        factory.setIdleEventInterval(60000L);
        ...

        return factory;
    }
In each of these cases, an event is published once per minute while the container is idle.

 

Event Consumption

您可以通过实现ApplicationListener来捕获空闲事件——要么是一个普通的侦听器，要么是一个缩小到只接收此特定事件的侦听器。您还可以使用在Spring Framework 4.2中引入的@EventListener。

下面的示例将@RabbitListener和@EventListener合并到一个类中。您需要了解应用程序侦听器获取所有容器的事件，因此，如果希望根据哪个容器空闲采取特定的操作，可能需要检查侦听器ID。您还可以为此使用@EventListener条件。

事件有四个属性:

•source:侦听器容器实例

•id:侦听器id(或容器bean名称)

•queueNames:容器侦听的队列的名称

下面的例子展示了如何使用@RabbitListener和@EventListener注解创建监听器:

 

public class Listener {

        @RabbitListener(id="someId", queues="#{queue.name}")
        public String listen(String foo) {

            return foo.toUpperCase();
        }

        @EventListener(condition = "event.listenerId == 'someId'")
        public void onApplicationEvent(ListenerContainerIdleEvent event) {

... }

}

 

 

事件监听器查看所有容器的事件。因此，在前面的示例中，我们根据侦听器ID缩小接收的事件。

如果希望使用idle事件来停止lister容器，则不应在调用侦听器的线程上调用container.stop()。这样做总是会导致延迟和不必要的日志消息。相反，应该将事件传递给另一个线程，该线程可以停止容器。