RabbitMQ 高级特性——事务

前言

前面我们学习了 RabbitMQ 的延迟队列，通过延迟队列可以实现生产者生产的消息不是立即被消费者消费。那么这篇文章我们将来学习 RabbitMQ 的事务。

事务

RabbitMQ 是基于 AMQP 协议实现的，该协议实现了事务机制，因此 RabbitMQ 也支持事务机制。Spring AMQP 也提供了对事务相关的操作。RabbitMQ 事务允许开发者确保消息的发送和接收是原子性的，要么全部成功，要么全部失败。然而，需要明确的是，RabbitMQ 的事务支持主要集中在生产者（发送方）端，并且它可能不是处理高并发场景下的最佳实践，因为使用事务会增加消息发送的延迟和复杂性。

那么我们来看看在 Spring 中如何实现 RabbitMQ 事务。

先来看看在没有事务的情况下是否能够保证消息发送的原子性：

public static final String TRANS_EXCHANGE = "trans.exchange";
public static final String TRANS_QUEUE = "trans.queue";

声明交换机、队列和绑定关系：

@Configuration
public class TransConfig {@Bean("transExchange")public Exchange transExchange() {return ExchangeBuilder.directExchange(Constants.TRANS_EXCHANGE).build();}@Bean("transQueue")public Queue transQueue() {return QueueBuilder.durable(Constants.TRANS_QUEUE).build();}@Bean("transBinding")public Binding transBinding(@Qualifier("transExchange") Exchange exchange, @Qualifier("transQueue") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("trans").noargs();}
}

消费者代码：

@RequestMapping("/trans")
public String trans() {rabbitTemplate.convertAndSend(Constants.TRANS_EXCHANGE,"trans","rabbitmq trans1");//制造异常int ret = 3/0;rabbitTemplate.convertAndSend(Constants.TRANS_EXCHANGE,"trans","rabbitmq trans2");return "消息发送成功";
}

这里就不指定消费者了，只是看看事务的效果。

观察队列的情况：

可以发现此时消息的发送是不具备原子性的，所以我们就使用事务保证消息的原子性。

配置事务管理器

@Configuration
public class TransConfig {@Bean("transactionManager")public RabbitTransactionManager transactionManager(ConnectionFactory factory) {return new RabbitTransactionManager(factory);}@Bean("transactionRabbitTemplate")public RabbitTemplate transactionRabbitTemplate(ConnectionFactory factory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(factory);rabbitTemplate.setChannelTransacted(true);return rabbitTemplate;}
}

加上@Transactional注解

配置完成事务管理器之后，我们需要在需要开启事务的方法上加上 @Transactional 注解：

@RequestMapping("producer")
@RestController
public class ProducerController {@Autowiredprivate RabbitTemplate rabbitTemplate;@Resource(name = "transactionRabbitTemplate")private RabbitTemplate transactionRabbitTemplate;@Transactional@RequestMapping("/trans")public String trans() {transactionRabbitTemplate.convertAndSend(Constants.TRANS_EXCHANGE,"trans","rabbitmq trans1");//制造异常int ret = 3/0;transactionRabbitTemplate.convertAndSend(Constants.TRANS_EXCHANGE,"trans","rabbitmq trans2");return "消息发送成功";}
}

上面就是 RabbitMQ 事务的使用。