52.RocketMQ 事务

今天的博客有点多，因为前几天一直用笔记录，今天都补上了。后面的博客先停一段时间，后面还有dubbo、storm、kafka、solor、nginx、keepalived、fastdfs等内容，只是因为最近准备跳槽，停更一段时间，等到新公司后再继续更新。

场景1：支付宝转1w到余额宝，支付宝扣了1w，服务挂了怎么办？余额还没有加上

场景2：订单系统和库存系统如何保持一致

如果是本地的话很好解决

begin transaction
　　update 支付宝 - 1w;
　　update 余额宝 + 1W;
end transaction
用Spring的话，方法上加 @Transaction注释

那如果是跨系统的呢？该如何解决？

有一种思路是这样的：

client发送转账请求给事务协调器
事务协调器先发送扣款请求给支付宝，返回执行结果（这时并没有提交）
事务协调器在发送加款请求给余额宝，返回执行结果（这时也没有提交）
事务协调器看两个执行结果都返回OK 就执行第四步，提交2和3没有提交的更新请求。

但是这个有个问题，那就是性能很受影响，主要卡在事务协调器这里。

RocketMQ的实现方式如下（图片来自网络）：

支付宝先生成扣款信息 --> 消息队列 --> 余额宝消费消息

发送消息：

 import com.alibaba.rocketmq.client.exception.MQClientException;

 import com.alibaba.rocketmq.client.producer.SendResult;

 import com.alibaba.rocketmq.client.producer.TransactionCheckListener;

 import com.alibaba.rocketmq.client.producer.TransactionMQProducer;

 import com.alibaba.rocketmq.common.message.Message;

 /**

  * 发送事务消息例子

  *

  */

 public class TransactionProducer {

     public static void main(String[] args) throws MQClientException, InterruptedException {

         TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();

         TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");

         // 事务回查最小并发数

         producer.setCheckThreadPoolMinSize(2);

         // 事务回查最大并发数

         producer.setCheckThreadPoolMaxSize(2);

         // 队列数

         producer.setCheckRequestHoldMax(2000);

         producer.setTransactionCheckListener(transactionCheckListener);

         producer.start();

         String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE" };

         TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();

         for (int i = 0; i < 100; i++) {

             try {

                 Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,

                     ("Hello RocketMQ " + i).getBytes());

                 SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);

                 System.out.println(sendResult);

                 Thread.sleep(10);

             }

             catch (MQClientException e) {

                 e.printStackTrace();

             }

         }

         for (int i = 0; i < 100000; i++) {

             Thread.sleep(1000);

         }

         producer.shutdown();

     }

 }

执行本地事务

 import java.util.concurrent.atomic.AtomicInteger;

 import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;

 import com.alibaba.rocketmq.client.producer.LocalTransactionState;

 import com.alibaba.rocketmq.common.message.Message;

 /**

  * 执行本地事务

  */

 public class TransactionExecuterImpl implements LocalTransactionExecuter {

     private AtomicInteger transactionIndex = new AtomicInteger(1);

     @Override

     public LocalTransactionState executeLocalTransactionBranch(final Message msg, final Object arg) {

         int value = transactionIndex.getAndIncrement();

         if (value == 0) {

             throw new RuntimeException("Could not find db");

         }

         else if ((value % 5) == 0) {

             return LocalTransactionState.ROLLBACK_MESSAGE;

         }

         else if ((value % 4) == 0) {

             return LocalTransactionState.COMMIT_MESSAGE;

         }

         return LocalTransactionState.UNKNOW;

     }

 }

服务器回查客户端（这个功能在开源版本中已经被咔掉了，但是我们还是要写，不然报错）

 import java.util.concurrent.atomic.AtomicInteger;

 import com.alibaba.rocketmq.client.producer.LocalTransactionState;

 import com.alibaba.rocketmq.client.producer.TransactionCheckListener;

 import com.alibaba.rocketmq.common.message.MessageExt;

 /**

  * 未决事务，服务器回查客户端

  */

 public class TransactionCheckListenerImpl implements TransactionCheckListener {

     private AtomicInteger transactionIndex = new AtomicInteger(0);

     @Override

     public LocalTransactionState checkLocalTransactionState(MessageExt msg) {

         System.out.println("server checking TrMsg " + msg.toString());

         int value = transactionIndex.getAndIncrement();

         if ((value % 6) == 0) {

             throw new RuntimeException("Could not find db");

         }

         else if ((value % 5) == 0) {

             return LocalTransactionState.ROLLBACK_MESSAGE;

         }

         else if ((value % 4) == 0) {

             return LocalTransactionState.COMMIT_MESSAGE;

         }

         return LocalTransactionState.UNKNOW;

     }

 }

到这就完了，为什么只介绍发送不介绍接收呢？因为一旦消息提交到MQ就不用管了，要相信MQ会把消息送达consumer，如果消息未能被成功消费的话，那么Producer也会回滚

如何保证分布式系统的全局性事务？

因为阿里在3.2.6版本后，砍掉了消息回查的功能，也就是consumer端是否成功消费，Producer端并不知道，所以如果要保证全局性事务，我们要有自己的实现机制：

巴特西

52.RocketMQ 事务

最新文章

热门文章