如何在使用@RabbitListener时配置消息监听容器参数？

一、核心思路 #

@RabbitListener通过RabbitListenerContainerFactory自动创建监听容器，所有容器级配置都需要通过自定义这个工厂Bean来实现，包括事务开启、并发数、预取数等各类参数。

二、配置事务行为 #

1. 自定义容器工厂并开启事务 #

创建SimpleRabbitListenerContainerFactory（或DirectRabbitListenerContainerFactory）的Bean，在其中配置事务管理器、开启channel事务：

@Configuration
public class RabbitListenerConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory rabbitConnectionFactory,
            PlatformTransactionManager transactionManager) {
        
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(rabbitConnectionFactory);
        
        // 开启channel事务
        factory.setChannelTransacted(true);
        // 指定事务管理器（若需联动数据库事务，可使用链式事务管理器）
        factory.setTransactionManager(transactionManager);
        
        return factory;
    }

    // 多资源事务配置（RabbitMQ + 数据库）
    @Bean
    public PlatformTransactionManager chainedTransactionManager(
            RabbitTransactionManager rabbitTxManager,
            DataSourceTransactionManager jdbcTxManager) {
        return new ChainedTransactionManager(rabbitTxManager, jdbcTxManager);
    }
}

2. 关联@RabbitListener到自定义工厂 #

如果仅定义了一个自定义工厂，它会成为默认工厂，@RabbitListener无需额外配置即可使用。若存在多个工厂，可通过containerFactory属性指定目标工厂：

@RabbitListener(queues = "some.queue", containerFactory = "rabbitListenerContainerFactory")
public void handleMessage(String message) {
    // 消息处理逻辑将运行在事务上下文内
}

三、配置其他容器选项 #

所有容器配置参数（如并发消费者数、预取数、重试策略等）都可以在容器工厂中统一设置，示例如下：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
        ConnectionFactory rabbitConnectionFactory,
        PlatformTransactionManager transactionManager) {
    
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(rabbitConnectionFactory);
    
    // 事务配置
    factory.setChannelTransacted(true);
    factory.setTransactionManager(transactionManager);
    
    // 并发消费者配置
    factory.setConcurrentConsumers(3);
    factory.setMaxConcurrentConsumers(5);
    
    // 预取数配置
    factory.setPrefetchCount(10);
    
    // 重试策略（需引入spring-retry依赖）
    RetryTemplate retryTemplate = new RetryTemplate();
    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(3);
    retryTemplate.setRetryPolicy(retryPolicy);
    factory.setRetryTemplate(retryTemplate);
    
    // 错误处理：指定致命异常判定规则，失败时转发死信
    factory.setErrorHandler(new ConditionalRejectingErrorHandler(
            new FatalExceptionStrategy() {
                @Override
                public boolean isFatal(Throwable t) {
                    return t instanceof AmqpRejectAndDontRequeueException;
                }
            }
    ));
    
    return factory;
}

四、方法级事务补充 #

如果需要更细粒度的事务控制，可在监听方法上直接添加@Transactional注解，此时需将容器的channelTransacted设为false，由Spring声明式事务接管：

@RabbitListener(queues = "some.queue")
@Transactional
public void handleMessage(String message) {
    // 数据库操作与Rabbit消息确认将在同一事务中执行
}

内容的提问来源于stack exchange，提问作者TRaether