快捷導(dǎo)航

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

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

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

一、RocketMq事務(wù)消息流程：

1、首先會(huì)向broker發(fā)送一個(gè)預(yù)請求消息，消費(fèi)者不可見

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

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

二、RocketMq事務(wù)消息實(shí)例：

1、引入rocketMq相關(guān)的依賴：

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

2、創(chuàng)建一個(gè)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í)行本地事務(wù)和消息回查
    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、啟動(dòng)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ā)送事務(wù)消息
    TransactionSendResult result = producer.sendMessageInTransaction(message, "hello-transaction");

    producer.shutdown();
  }
}

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

public class TransactionListenerImol implements TransactionListener {
  //存儲(chǔ)事務(wù)狀態(tài)信息 key:事務(wù)id value：當(dāng)前事務(wù)執(zhí)行的狀態(tài)
  private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
  //執(zhí)行本地事務(wù)
  @Override
  public LocalTransactionState executeLocalTransaction(Message message, Object o) {
    //事務(wù)id
    String transactionId = message.getTransactionId();
    //0:執(zhí)行中，狀態(tài)未知 1：執(zhí)行成功 2：執(zhí)行失敗
    localTrans.put(transactionId, 0);
    //業(yè)務(wù)執(zhí)行，本地事務(wù)，service
    System.out.println("hello-demo-transaction");
    try {
      System.out.println("正在執(zhí)行本地事務(wù)---");
      Thread.sleep(60000*2);
      System.out.println("本地事務(wù)執(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) {
    //獲取對應(yīng)事務(wù)的狀態(tài)信息
    String transactionId = messageExt.getTransactionId();
    //獲取對應(yīng)事務(wù)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)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: