RocketMq事務(wù)消息發(fā)送代碼的過(guò)程詳解

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

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)聽(tīng)對(duì)象用于執(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)建消息對(duì)象，指定主題Topic、Tag和消息體 String topic, String tags, String keys, byte[] body
    Message message = new Message("Topic_transaction_demo", //主題
        "Tags", //主要用于消息過(guò)濾
        "Key_1", //消息唯一值
        ("hello-transaction").getBytes(RemotingHelper.DEFAULT_CHARSET));

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

    producer.shutdown();
  }
}

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) {
    //獲取對(duì)應(yīng)事務(wù)的狀態(tài)信息
    String transactionId = messageExt.getTransactionId();
    //獲取對(duì)應(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;
  }
}