事务消息的使用场景
事务消息的使用场景很多，比如在电商系统中用户下单后新增了订单记录，对应的商品库存需要减少。怎么保证新增订单后商品库存减少呢？又例如红包业务，张三给李四发红包，张三的账户余额需要扣减，李四的账户余额需要增加，怎么保证张三账户扣减李四账户加钱呢？
这些问题都是事务问题，可以简单理解为：一个表的数据更新后，如何保证另外一个表的数据也更新成功。如果使用同一个数据库实例，那么问题很简单，可以使用本地事务来解决，Spring的@Transactional注解就支持。
但实际场景并不这么简单，互联网应用的流量大，系统规模通常也比较大，会存在许多数据库实例、分库分表等。我们需要修改的表往往不在同一个数据库实例或者同一个数据库中，此时就不能使用本地事务来解决了，需要用到分布式事务。RocketMQ的一大特点就是支持事务消息，支持一些分布式事务场景。
如何发送事务消息
我们使用RocketMQ事务消息来模拟下单减库存的场景，代码仅包含核心功能代码
1. 发送订单的事务消息，预提交
@RestControllerpublic TransactionalController { @Autowired private Source source; public String transactional() { Order order = new Order("123", "test"); String transactionId = UUID.randomUUID().toString(); MessageBuilder builder = MessageBuilder.withPayload(order).setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId); Message message = builder.build(); source.output().send(message); return "OK"; }}server.port=8081spring.cloud.stream.rocketmq.binder.name-server=192.168.0.118:9876spring.cloud.stream.bindings.output.destination=TopicTestspring.cloud.stream.rocketmq.bindings.output.producer.group=demo-groupOrder对象保存了订单信息，随机生成一个ID作为消息的事务ID。此时消息已经发送到Broker中，但还未投递出去，Consumer暂时还不能消费这条消息。
2. 执行订单信息入库的事务操作，提交或回滚事务消息
@RocketMQTransactionListener(txProducerGroup = "OrderTransactionGroup")public TransactionalMsgListener implements RocketMQLocalTransactionListener { @Override public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) { try { // 获取前面生成的事务ID String transactionId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID); // 以事务ID为主键，执行本地事务 Order order = (Order) message.getPayload(); boolean result = this.saveOrder(order, transactionId); return result ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK; } catch (Exception e) { return RocketMQLocalTransactionState.ROLLBACK; } } private boolean saveOrder(Order order, String transactionId) { // 将事务ID设置为唯一键 // 调用数据库Insert into 订单表 return true; } @Override public RocketMQLocalTransactionState checkLocalTransaction(Message message) { // 获取事务ID String transactionId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID); // 以事务ID为主键，查询本地事务执行情况 if (isSuccess(transactionId)) { return RocketMQLocalTransactionState.COMMIT; } return RocketMQLocalTransactionState.ROLLBACK; } private boolean isSuccess(String transactionId) { // 查询数据库 select from 订单表 return true; }}实现RocketMQLocalTransactionListener接口，使用@RocketMQTransactionListener注解用于接收本地事务的监听，txProducerGroup是事务组名称，RocketMQLocalTransactionListener接口有两个实现方法：
executeLocalTransaction：执行本地事务，在第一步中消息发送成功会回调执行，一旦事务提交成功，下游应用的Consumer能收到该消息，在这里demo的本地事务就是保存订单信息入库checkLocalTransaction：检查本地事务执行状态，如果executeLocalTransaction方法中返回的状态是未知UNKNOWN或者未返回状态，默认会在预处理发送的1分钟后由Broker通知Producer检查本地事务，在Producer中回调本地事务监听器中的checkLocalTransaction方法。检查本地事务时，可以根据事务ID查询本地事务的状态，再返回具体事务状态给Broker。3. 消费订单消息
@EnableBinding({Sink.class})@SpringBootApplicationpublic App { public static void main( String[] args ) { SpringApplication.run(App.class); } @StreamListener(Sink.INPUT) public void receive(String msg) { System.out.println("TopicTest receive: " + msg + ", receiveTime= " + System.currentTimeMillis()); }}server.port=8091spring.cloud.stream.rocketmq.binder.name-server=192.168.0.118:9876spring.cloud.stream.bindings.input.destination=TopicTestspring.cloud.stream.bindings.input.group=test-group1消费事务消息与消费普通消息的代码是一样的，无需做任何修改。
事务消息的技术原理
RocketMQ采用了2PC的方案来提交事务消息。第一阶段Producer向Broker发送预处理消息，此时消息还未被投递出去，Consumer还不能消费；第二阶段Producer向Broker发送提交或者回滚消息。具体流程如下：
发送预处理消息成功后，开始执行本地事务。如果本地事务执行成功，发送提交请求提交事务消息，消息会投递给Consumer
如果本地事务执行失败，发送回滚请求回滚事务消息，消息不会投递给Consumer
如果本地事务状态未知，网络故障或Producer宕机，Broker未收到二次确认的消息。由Broker端发送请求给Producer进行消息回查，确认提交或回滚。如果消息状态一致未被确认，需要人工介入处理。 