java - 哪种类型的 Spring 集成 channel ？

Question

我在 ActiveMQ 中有一个队列，希望使用 spring 集成将消息从队列中取出到我们的应用程序中。我们将应用程序部署了两次(以防其中一个失败)——每条消息只能由其中一个应用程序处理。此外，我需要一个 jms tx 管理器，以防在消息处理期间出现致命的应用程序故障。因此，我的 channel 适配器如下所示:

<int-jms:message-driven-channel-adapter 
    channel="myChannel" 
    connection-factory="jmsConnectionFactory"
    pub-sub-domain="false"
    destination-name="MY_QUEUE" 
    transaction-manager="jmsTxManager" />

当 myChannel 是直接 channel 时，这一切都工作正常，但是我想使用任务执行器，以便可以一次处理许多消息。

为了应对致命的应用程序故障，我认为集合 channel 可能是一种可行的方法(我相信当任务执行器中的线程空闲时， channel 适配器将转到 Activity mq 以检索另一条消息)， channel 上的内存中不会保存任何消息。这似乎不起作用，以下代码抛出 TaskRejectException:

<int:channel id="myChannel">
    <int:rendezvous-queue />
</int:channel>
<task:executor id="taskExecutor" pool-size="2" queue-capacity="0" />
<int:router input-channel="myChannel" expression="payload.getType() + 'Channel'">
    <int:poller fixed-rate="1000" task-executor="taskExecutor" />
</int:router>

位于路由器之后的服务激活器同步处理，需要 10 秒来处理，所以我预计在 t=0s 时会消失并检索 2 条消息(线程池的大小)，处理它们，释放在 t=10s 时将线程加入线程池，然后再次向 active mq 请求消息。然而，似乎在 t=0s 时检索到了超过 2 条消息。

有人可以建议我应该做什么吗？

Answer 1

为了进行交易，您必须使用直接 channel 。

您可以使用消息驱动 channel 适配器上的并发属性(concurrent-consumers 和 max-concurrent-消费者)。