java - Spring AMQP Listener Container 中的并发是如何实现的？

Question

我的容器 XML 配置:

<rabbit:listener-container
        connection-factory="myConnectionFactory"
        acknowledge="none"
        concurrency="10"
        requeue-rejected="false">
    <rabbit:listener ref="myListener" queues="myQueue"/>
</rabbit:listener-container>

而 myListener 只是一个类

@Component("myListener")
public class MyListener implements MessageListener {
    @Autowired
    SomeDependency dependency;
    ....
}

我在我的 XML 中指定了 concurrency="10"。这确切地是什么意思？

---

我找到了some docs .它们没有那么有用的说明:

Specify the number of concurrent consumers to create. Default is 1.

---

我感兴趣的是 MyListener 是否必须是线程安全的，即

• 是否创建了多个实例或多个线程使用了单个实例？
• 我可以访问不同步的实例字段吗？
• 是 SomeDependency 依赖 实例化一次还是为每个线程/实例实例化？
• 依赖是否需要线程安全？

Answer 1

是的，要使用并发，你的监听器必须是线程安全的。每个容器有一个监听器实例。但是，<rabbit:listener-container/> namespace 元素实际上只是为了方便添加“共享”属性，每个监听器元素都有自己的容器。

通常最好使用无状态对象(没有写入的字段)，但这并不总是可行的。

如果你的监听器不是线程安全的，你可以使用...

<rabbit:listener-container
    connection-factory="myConnectionFactory"
    acknowledge="none"
    requeue-rejected="false">
    <rabbit:listener ref="myListener" queues="myQueue"/>
    <rabbit:listener ref="myListener" queues="myQueue"/>
    <rabbit:listener ref="myListener" queues="myQueue"/>
    <rabbit:listener ref="myListener" queues="myQueue"/>
    ...
</rabbit:listener-container>

...并添加 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) .然后，您将为每个监听器获取一个容器，并且将向每个监听器注入(inject)不同的监听器实例。

您还需要原型(prototype)作用域用于注入(inject)到监听器中的任何非线程安全依赖项。