分布式事務中使用RocketMQ的事務消息機制優(yōu)化事務的處理邏輯

支付寶賬戶表：A（id，userId，amount）
余額寶賬戶表：B（id，userId，amount）
用戶的userId=1；

1）支付寶表扣除1萬：update A set amount=amount-10000 where userId=1;
2）余額寶表增加1萬：update B set amount=amount+10000 where userId=1;

Begin transaction 
  update A set amount=amount-10000 where userId=1;
 update B set amount=amount+10000 where userId=1;
End transaction 
commit;

@Transactional(rollbackFor=Exception.class) 
public void update() { 
      //更新A表 
      updateATable();
      //更新B表
      updateBTable();
}

1）兩階段提交涉及多次節(jié)點間的網絡通信，通信時間太長！
2）事務時間相對于變長了，鎖定的資源的時間也變長了，造成資源等待時間也增加好多！

Begin transaction 
       update A set amount=amount-10000 where userId=1; 
       insert into message(userId, amount,status) values(1, 10000, 1); 
End transaction 
commit;

For each msg in queue 
 Begin transaction 
   select count(*) as cnt from message_apply where msg_id=msg.msg_id; 
   if cnt==0 then 
  update B set amount=amount+10000 where userId=1; 
  insert into message_apply(msg_id) values(msg.msg_id); 
   end if
 End transaction 
 commit;
End For

// =============================發(fā)送事務消息的一系列準備工作========================================
// 未決事務，MQ服務器回查客戶端
// 也就是上文所說的，當RocketMQ發(fā)現`Prepared消息`時，會根據這個Listener實現的策略來決斷事務
TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
// 構造事務消息的生產者
TransactionMQProducer producer = new TransactionMQProducer("groupName");
// 設置事務決斷處理類
producer.setTransactionCheckListener(transactionCheckListener);
// 本地事務的處理邏輯，相當于示例中檢查Bob賬戶并扣錢的邏輯
TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
producer.start()
// 構造MSG，省略構造參數
Message msg = new Message(......);
// 發(fā)送消息
SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
producer.shutdown();

//  ================================事務消息的發(fā)送過程=============================================
public TransactionSendResult sendMessageInTransaction(.....) {
    // 邏輯代碼，非實際代碼
    // 1.發(fā)送消息
    sendResult = this.send(msg);
    // sendResult.getSendStatus() == SEND_OK
    // 2.如果消息發(fā)送成功，處理與消息關聯的本地事務單元
    LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
    // 3.結束事務
    this.endTransaction(sendResult, localTransactionState, localException);
}
 

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("transaction_producer");
        //設置用于事務消息的處理線程池
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setExecutorService(executorService);
        //設置事務監(jiān)聽器，監(jiān)聽器實現接口org.apache.rocketmq.client.producer.TransactionListener
        //監(jiān)聽器中實現需要處理的交易業(yè)務邏輯的處理，以及MQ Broker中未確認的事務與業(yè)務的確認邏輯
        producer.setTransactionListener(transactionListener);
        producer.start();
 
        //生成不同的Tag，用于模擬不同的處理場景
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
             //組裝產生消息
                Message msg =
                    new Message("TopicTransaction", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                //以事務發(fā)送消息，并在事務消息被成功預寫入到RocketMQ中后，執(zhí)行用戶定義的交易邏輯，
                //交易邏輯執(zhí)行成功后，再實現實現業(yè)務消息的提交邏輯
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);
                System.out.printf("%s%n", sendResult.getTransactionId());
                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
 
        producer.shutdown();
    }
}

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
 
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
 
    /**
     * 該方法會在消息成功預寫入RocketMQ后被執(zhí)行
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        System.out.println("開始處理業(yè)務邏輯...");
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        switch (status) {
         case 0:
          //LocalTransactionState.UNKNOW表示未知的事件，需要RocketMQ進一步服務業(yè)務進行確認該交易的處理
          //結果，確認消息被調用的方法為下方的checkLocalTransaction。
          //注：RocketMQ與業(yè)務確認消息的執(zhí)行狀態(tài)的功能已經被移除了，在早期3.0.8的版本中有該功能，因而如果
          //返回的狀態(tài)為UNKNOW，則該消息不會被提交
          return LocalTransactionState.UNKNOW;
         case 1:
          return LocalTransactionState.COMMIT_MESSAGE;
         case 2:
          return LocalTransactionState.ROLLBACK_MESSAGE;
         default:
          return LocalTransactionState.COMMIT_MESSAGE;
        }
    }
 
    /**
     * 該方法用于RocketMQ與業(yè)務確認未提交事務的消息的狀態(tài)，不過該方法已經的實現在RocketMQ中已經
     * 被刪除了，因而其功能也就沒有意義了。
     * 不過如果使用阿里云的企業(yè)的RocketMQ服務，該功能會起作用。
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        int mod = msg.getTransactionId().hashCode() % 2;
        if (null != status) {
            switch (mod) {
                case 0:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                default:
                    return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

public class Consumer {
 public static void main(String[] args) throws MQClientException {
  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_producer");
  consumer.setNamesrvAddr("127.0.0.1:9876");
  /**
   * 設置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費<br>
   * 如果非第一次啟動，那么按照上次消費的位置繼續(xù)消費
   */
  consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  consumer.subscribe("TopicTransaction", "*");
  consumer.registerMessageListener(new MessageListenerOrderly() {
   private Random random = new Random();
 
   @Override
   public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
    // 設置自動提交
    context.setAutoCommit(true);
    for (MessageExt msg : msgs) {
     System.out.println("獲取到消息開始消費："+msg + " , content : " + new String(msg.getBody()));
    }
    try {
     // 模擬業(yè)務處理
     TimeUnit.SECONDS.sleep(random.nextInt(5));
    } catch (Exception e) {
     e.printStackTrace();
     //返回處理失敗，該消息后續(xù)可以繼續(xù)被消費
     return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
    }
    //返回處理成功，該消息就不會再次投遞過來了
    return ConsumeOrderlyStatus.SUCCESS;
   }
  });
  consumer.start();
  System.out.println("consumer start ! ");
 }
}