spring boot集成rabbitmq实现事务控制+手动确认

参考博客

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: xxx
    password: xxx
    #virtual-host: xxx
    # 确认消息已发送到交换机,若发送失败则回调confirmCallback方法
    # 开启事务机制就不能开启消息确认机制
    #publisher-confirm-type: correlated
    # 确认消息已发送到队列,若发送失败则回调returnCallback方法
    publisher-returns: true
    # 手动确认消息
    listener:
      simple:
        acknowledge-mode: manual
      direct:
        acknowledge-mode: manual

事务控制 + 消息回调

@Slf4j
@Configuration
public class MQConfig implements RabbitListenerConfigurer {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //消息序列化
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter(new ObjectMapper()));
        //开启事务,只需要事务的话该函数剩下的代码可不要
        rabbitTemplate.setChannelTransacted(true);
        
        //交换器无法根据自身类型和路由键找到一个符合条件的队列时的处理方式
        //true:将消息返回给生产者   false:直接丢弃
        rabbitTemplate.setMandatory(true);

        //消息是否到交换机的回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean isAck, String s) {
                if (isAck) {
                    log.info("消息已发送到交换器 cause:{} - {}", s, correlationData.toString());
                } else {
                    log.info("消息未发送到交换器 cause:{} - {}", s, correlationData.toString());
                }

            }
        });
        //消息没有到队列则回调该函数
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.info("消息被退回 {}", returnedMessage.toString());
            }
        });

        return rabbitTemplate;
    }

    //消息反序列化
    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
    }

    @Bean
    MessageHandlerMethodFactory messageHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
        messageHandlerMethodFactory.setMessageConverter(mappingJackson2MessageConverter());
        return messageHandlerMethodFactory;
    }

    @Bean
    public MappingJackson2MessageConverter mappingJackson2MessageConverter() {
        return new MappingJackson2MessageConverter();
    }


    //rabbitmq事务管理器
    @Bean("rabbitTransactionManager")
    RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }
}

指明事务管理器的名称

    @Transactional(transactionManager = "rabbitTransactionManager")
    public void transactionAsyncReduceStock() {
        //业务逻辑
        rabbitTemplate.convertAndSend("交换机名字", "route-key", 数据);
    }

手动确认

	@RabbitListener(queues = "队列名字")
    @RabbitHandler
    public void miaosha(Message message, Channel channel) throws IOException {
        log.info("接收到的数据为 ====> {}",message.getbody());
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        

    }