52.RocketMQ 事務

今天的博客有點多，因為前幾天一直用筆記錄，今天都補上了。后面的博客先停一段時間，后面還有dubbo、storm、kafka、solor、nginx、keepalived、fastdfs等內容，只是因為最近准備跳槽，停更一段時間，等到新公司后再繼續更新。

場景1：支付寶轉1w到余額寶，支付寶扣了1w，服務掛了怎么辦？余額還沒有加上

場景2：訂單系統和庫存系統如何保持一致

如果是本地的話很好解決

• begin transaction
  　　update 支付寶 - 1w;
  　　update 余額寶 + 1W;
  end transaction
• 用Spring的話，方法上加 @Transaction注釋

那如果是跨系統的呢？該如何解決？

有一種思路是這樣的：

1.  client發送轉賬請求給事務協調器
2. 事務協調器先發送扣款請求給支付寶，返回執行結果（這時並沒有提交）
3. 事務協調器在發送加款請求給余額寶，返回執行結果（這時也沒有提交）
4. 事務協調器看兩個執行結果都返回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端並不知道，所以如果要保證全局性事務，我們要有自己的實現機制：