异步消费者

Spring AMQP 还支持通过使用 @RabbitListener 注解实现注解驱动的监听器端点,并提供了开放的基础设施以编程方式注册端点。这是目前为止设置异步消费者最便捷的方式。有关更多详细信息,请参阅 注解驱动的监听器端点

prefetch 的默认值以前是 1,这可能导致高效消费者的利用率不足。从 2.0 版本开始,默认的 prefetch 值现在是 250,这应该能在大多数常见场景下保持消费者忙碌,从而提高吞吐量。

然而,在某些场景下,prefetch 值应该设置得较低

  • 对于大型消息,特别是处理缓慢的情况下(消息可能会在客户端进程中占用大量内存)

  • 当需要严格的消息顺序时(在这种情况下,prefetch 值应重新设置为 1)

  • 其他特殊情况

此外,在消息量低且消费者数量多(包括单个监听器容器实例内的并发)的情况下,您可能希望降低 prefetch 值,以便在消费者之间实现更均匀的消息分发。

关于 prefetch 的更多背景信息,请参阅这篇关于 RabbitMQ 中的消费者利用率 的文章,以及这篇关于 排队论 的文章。

消息监听器

对于异步 Message 接收,涉及一个专门的组件(而不是 AmqpTemplate)。该组件是一个用于 Message 消费回调的容器。我们将在本节后面讨论容器及其属性。但是,首先我们应该看一下回调,因为这是应用程序代码与消息系统集成的地方。回调有几种选择,首先是 MessageListener 接口的实现,如下所示

public interface MessageListener {
    void onMessage(Message message);
}

如果您的回调逻辑因任何原因依赖于 AMQP Channel 实例,则可以改用 ChannelAwareMessageListener。它看起来类似,但有一个额外的参数。以下列表显示了 ChannelAwareMessageListener 接口的定义

public interface ChannelAwareMessageListener {
    void onMessage(Message message, Channel channel) throws Exception;
}
在 2.1 版本中,此接口从包 o.s.amqp.rabbit.core 移至 o.s.amqp.rabbit.listener.api

MessageListenerAdapter

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

1.5 版本引入了一种更灵活的 POJO 消息机制,即 @RabbitListener 注解。有关更多信息,请参阅 注解驱动的监听器端点

使用适配器时,您只需提供适配器本身应调用的实例引用。以下示例展示了如何实现

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

您可以继承适配器并提供 getListenerMethodName() 的实现,以根据消息动态选择不同的方法。此方法有两个参数:originalMessageextractedMessage,后者是任何转换的结果。默认情况下,配置了一个 SimpleMessageConverter。有关更多信息以及其他可用转换器的信息,请参阅 SimpleMessageConverter

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

从 1.5 版本开始,您可以配置一个将消费者队列或标签映射到方法名的 Map,以动态选择要调用的方法。如果 Map 中没有对应的条目,则回退到默认监听器方法。默认监听器方法(如果未设置)为 handleMessage

从 2.0 版本开始,提供了一个方便的 FunctionalInterface。以下列表显示了 FunctionalInterface 的定义

@FunctionalInterface
public interface ReplyingMessageListener<T, R> {

    R handleMessage(T t);

}

该接口通过使用 Java 8 lambda 表达式简化了适配器的配置,如下例所示

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

从 2.2 版本开始,buildListenerArguments(Object) 方法已被弃用,取而代之的是新的 buildListenerArguments(Object, Channel, Message) 方法。新方法帮助监听器获取 ChannelMessage 参数以便执行更多操作,例如在手动确认模式下调用 channel.basicReject(long, boolean)。以下列表显示了最基本的示例

public class ExtendedListenerAdapter extends MessageListenerAdapter {

    @Override
    protected Object[] buildListenerArguments(Object extractedMessage, Channel channel, Message message) {
        return new Object[]{extractedMessage, channel, message};
    }

}

现在,如果您需要接收“channel”和“message”参数,可以将 ExtendedListenerAdapter 配置得与 MessageListenerAdapter 相同。监听器的参数应设置为 buildListenerArguments(Object, Channel, Message) 的返回值,如下面的监听器示例所示

public void handleMessage(Object object, Channel channel, Message message) throws IOException {
    ...
}

容器

现在您已经了解了 Message 监听回调的各种选项,我们可以将注意力转向容器。基本上,容器处理“主动”的职责,以便监听器回调可以保持被动。容器是一个“生命周期”组件的示例。它提供了启动和停止的方法。配置容器时,您实际上是在连接 AMQP 队列和 MessageListener 实例。您必须提供对 ConnectionFactory 以及该监听器应从中消费消息的队列名称或 Queue 实例的引用。

在 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>

以下列表展示了使用 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 CachingConnectionFactory 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);
            }
        };
    }
}

消费者优先级

从 RabbitMQ 3.2 版本开始,代理现在支持消费者优先级(请参阅 在 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 队列

当容器配置为监听 auto-delete 队列、队列设置了 x-expires 选项,或者在代理上配置了 Time-To-Live 策略时,当容器停止时(即最后一个消费者取消时),队列会被代理移除。在 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" />

在这种情况下,队列和交换器由 containerAdmin 声明,该 admin 设置了 auto-startup="false",因此元素在上下文初始化期间不会被声明。同样,容器也不会因为同样的原因而启动。当容器稍后启动时,它会使用对 containerAdmin 的引用来声明这些元素。