RocketMq事務消息發(fā)送代碼流程詳解

更新時間：2020年07月17日 09:16:56 作者：杯莫停、

這篇文章主要介紹了RocketMq事務消息發(fā)送代碼流程詳解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

一、RocketMq事務消息流程：

1、首先會向broker發(fā)送一個預請求消息，消費者不可見

2、回調(diào)執(zhí)行本地事務（比如操作數(shù)據(jù)庫）

3、事務執(zhí)行成功后，再次發(fā)送消息給broker，告訴broker事務執(zhí)行成功這個消息要提交，讓消費者可見。如果本地事務執(zhí)行超時，會返回一個unknow，broker會發(fā)送一個消息回查，檢查消息是否執(zhí)行成功。

二、RocketMq事務消息實例：

1、引入rocketMq相關的依賴：

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client</artifactId>
  <version>4.4.0</version>
</dependency>

2、創(chuàng)建一個TransactionProducer類：

public class TransactionProducer {

  public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException {
    //創(chuàng)建生產(chǎn)者并制定組名
    TransactionMQProducer producer = new TransactionMQProducer("rocketMQ_transaction_producer_group");
    //2.指定Nameserver地址
    producer.setNamesrvAddr("192.168.***.***:9876");
    //3、指定消息監(jiān)聽對象用于執(zhí)行本地事務和消息回查
    TransactionListener listener = new TransactionListenerImol();
    producer.setTransactionListener(listener);
    //4、線程池
    ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
      @Override
      public Thread newThread(Runnable r) {
        Thread thread = newThread(r);
        thread.setName("client-tanscation-msg-check-thread");
        return thread;
      }
    });
    producer.setExecutorService(executorService);
    //5、啟動producer
    producer.start();

    //6.創(chuàng)建消息對象，指定主題Topic、Tag和消息體 String topic, String tags, String keys, byte[] body
    Message message = new Message("Topic_transaction_demo", //主題
        "Tags", //主要用于消息過濾
        "Key_1", //消息唯一值
        ("hello-transaction").getBytes(RemotingHelper.DEFAULT_CHARSET));

    //7、發(fā)送事務消息
    TransactionSendResult result = producer.sendMessageInTransaction(message, "hello-transaction");

    producer.shutdown();
  }
}

3、發(fā)送事務消息還需要一個事務監(jiān)聽對象，它實現(xiàn)TransactionListener 接口，其中有兩個方法作用分別是執(zhí)行本地事務和消息回查：

public class TransactionListenerImol implements TransactionListener {
  //存儲事務狀態(tài)信息 key:事務id value：當前事務執(zhí)行的狀態(tài)
  private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
  //執(zhí)行本地事務
  @Override
  public LocalTransactionState executeLocalTransaction(Message message, Object o) {
    //事務id
    String transactionId = message.getTransactionId();
    //0:執(zhí)行中，狀態(tài)未知 1：執(zhí)行成功 2：執(zhí)行失敗
    localTrans.put(transactionId, 0);
    //業(yè)務執(zhí)行，本地事務，service
    System.out.println("hello-demo-transaction");
    try {
      System.out.println("正在執(zhí)行本地事務---");
      Thread.sleep(60000*2);
      System.out.println("本地事務執(zhí)行成功---");
      localTrans.put(transactionId, 1);
    } catch (InterruptedException e) {
      e.printStackTrace();
      localTrans.put(transactionId, 2);
      return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    return LocalTransactionState.COMMIT_MESSAGE;
  }

  //消息回查
  @Override
  public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
    //獲取對應事務的狀態(tài)信息
    String transactionId = messageExt.getTransactionId();
    //獲取對應事務id執(zhí)行狀態(tài)
    Integer status = localTrans.get(transactionId);
    //消息回查
    System.out.println("消息回查---transactionId:" + transactionId + "狀態(tài):" + status);
    switch (status) {
      case 0:
        return LocalTransactionState.UNKNOW;
      case 1:
        return LocalTransactionState.COMMIT_MESSAGE;
      case 2:
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    return LocalTransactionState.UNKNOW;
  }
}

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: