线程与异步消费者

Doubao 中英对照 Threading and Asynchronous Consumers

异步消费者会涉及若干不同的线程。

当 RabbitMQ Client 投递新消息时，会使用 SimpleMessageListenerContainer 中配置的 TaskExecutor 的线程来调用 MessageListener。如果未配置，则使用 SimpleAsyncTaskExecutor。如果你使用池化执行器，需要确保池大小足以处理所配置的并发度。对于 DirectMessageListenerContainer，MessageListener 直接在 RabbitMQ Client 线程上被调用。在这种情况下，taskExecutor 用于监控消费者的任务。

备注

当使用默认的 SimpleAsyncTaskExecutor 时，对于调用监听器的线程，监听容器的 beanName 会被用于 threadNamePrefix。这对日志分析很有用。我们通常建议在日志 appender 配置中始终包含线程名。当通过容器上的 taskExecutor 属性专门提供一个 TaskExecutor 时，它会原样使用、不做修改。建议你使用类似的方式为自定义 TaskExecutor bean 定义所创建的线程命名，以便在日志消息中识别线程。

在 CachingConnectionFactory 中配置的 Executor 会在创建连接时传入 RabbitMQ Client，它的线程用于把新消息投递给监听容器。如果未配置此项，客户端会为每个连接使用一个内部线程池执行器，（在撰写本文时）其池大小为 Runtime.getRuntime().availableProcessors() * 2。

如果你有大量工厂，或正在使用 CacheMode.CONNECTION，可能希望考虑使用一个共享的 ThreadPoolTaskExecutor，并配备足够的线程来满足你的工作负载。

important

对于 DirectMessageListenerContainer，你需要确保连接工厂配置了一个任务执行器，其线程数足以支持所有使用该工厂的监听容器所期望的并发度。默认池大小（在撰写本文时）为 Runtime.getRuntime().availableProcessors() * 2。

RabbitMQ client 使用一个 ThreadFactory 来为底层 I/O（socket）操作创建线程。要修改这个工厂，你需要配置底层的 RabbitMQ ConnectionFactory，如配置底层客户端连接工厂中所讨论。