spring集成：AmqpInboundChannelAdapter上的TaskExecutor和MaxConcurrentConsumers

Question

我的spring集成应用程序使用来自RabbitMQ的消息，将它们转换为SOAP消息，并执行web服务请求。

每秒可以从队列中获得许多（10-50）条消息。或者在应用程序重新启动后，RabbitMQ队列中可能有数千条消息。

在并行线程中处理多达10条消息的最佳可能场景是什么（消息排序很好，但不是必需的特性，如果web服务的回答是业务失败，那么失败的消息应该重试直到成功）。

Amqp侦听器不应该从队列中消耗更多的消息，因为任务执行器中可用的线程不忙。我可以在一个通道中定义一个ThreadExecutor,如下所示：

@Bean
public AmqpInboundChannelAdapterSMLCSpec 
amqpInboundChannelAdapter(ConnectionFactory connectionFactory, Queue queue) 
{
  return Amqp.inboundAdapter(connectionFactory, queue);
}

IntegrationFlow integrationFlow = IntegrationFlows
  .from(amqpInboundChannelAdapter)
  .channel(c -> c.executor(exportFlowsExecutor))
  .transform(businessObjectToSoapRequestTransformer)
  .handle(webServiceOutboundGatewayFactory.getObject())
  .get();

或者像这样在AmqpInboundChannelAdapter中定义一个任务执行器，而不在流定义中定义channels任务执行器就足够了：

@Bean
public AmqpInboundChannelAdapterSMLCSpec 
amqpInboundChannelAdapter(ConnectionFactory connectionFactory, Queue queue) 
{
  return Amqp.inboundAdapter(connectionFactory, queue)
             .configureContainer(c->c.taskExecutor(taskExecutor));
}

或者可以像选项1那样为通道定义任务执行器,但还可以在通道适配器上设置maxConcurrentConsumers,如下所示：

@Bean
public AmqpInboundChannelAdapterSMLCSpec 
amqpInboundChannelAdapter(ConnectionFactory connectionFactory, Queue queue) 
{
  return Amqp.inboundAdapter(connectionFactory, queue)
             .configureContainer(c->c.maxConcurrentConsumers(10));
}

Answer 1

在ListenerContainer上配置并发并让所有下游进程发生在来自容器的线程上的最佳实践。这样，当因为线程忙而没有更多的消息从队列中轮询时，您就会得到一个自然的背压。另一方面，由于在侦听器容器之后使用executorchannel释放一个轮询线程，并且当前消息将被当作已用消息进行处理，所以不会导致消息丢失，但是可能会在下游失败。