【RocketMQ系列六】RocketMQ事务消息

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_transaction_consumer");
    consumer.setNamesrvAddr("172.31.184.89:9876");
    consumer.subscribe("TransactionTopic", "*");
    // 4.创建一个回调函数
    consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
      // 5.处理消息
      for (MessageExt msg : msgs) {
        System.out.println(msg);
        System.out.println("收到的消息内容：" + new String(msg.getBody()));
      }
      // 返回消费成功的对象
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    // 6.启动消费者
    consumer.start();
    System.out.println("消费者已经启动");

public class TransactionListenerImpl implements TransactionListener {
  /**
   * 执行本地事务
   * 当事务half消息发送成功，这个方法将被执行
   * 事务的half消息是发到 RMQ_SYS_TRANS_OP_HALF_TOPIC 的topic中
   *
   * @param msg 消息
   * @param arg arg 自定义业务参数
   * @return {@link LocalTransactionState}
   */
  @Override
  public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    String tags = msg.getTags();
    System.out.println("============执行executeLocalTransaction方法，;消息内容是="+new String(msg.getBody()));
    if (StringUtils.contains(tags, "tagA")) {
      return LocalTransactionState.COMMIT_MESSAGE;
    } else if (StringUtils.contains(tags, "tagB")) {
      return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    return LocalTransactionState.UNKNOW;
  }
  /**
   * 检查本地事务
   * 回查本地事务状态，当half消息没响应时调用。
   *
   * @param msg 消息
   * @return {@link LocalTransactionState}
   */
  @Override
  public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    String tags = msg.getTags();
    System.out.println("============执行checkLocalTransaction方法，;消息内容是="+new String(msg.getBody()));
    if (StringUtils.contains(tags, "tagC")) {
      return LocalTransactionState.COMMIT_MESSAGE;
    } else if (StringUtils.contains(tags, "tagD")) {
      return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    return LocalTransactionState.UNKNOW;
  }
}

//1.定义事务监听器
    TransactionListener transactionListener = new TransactionListenerImpl();
    //2.定义生产者
    TransactionMQProducer producer = new TransactionMQProducer("transaction_produce_group");
    producer.setNamesrvAddr("172.31.184.89:9876");
    //3.定义线程池
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 10, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100), (runnable, executor) -> {
      BlockingQueue<Runnable> queue = executor.getQueue();
      try {
        queue.put(runnable);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    });
    //4.设置线程池
    producer.setExecutorService(threadPoolExecutor);
    //5.设置事务监听器
    producer.setTransactionListener(transactionListener);
    // 启动生产者
    producer.start();
    String[] tags = {"tagA", "tagB", "tagC", "tagD","tagE"};
    //发送10条half消息，消费者是收不到half消息的
    for (int i = 0; i < 10; i++) {
      Message message = new Message("TransactionTopic", tags[i % tags.length],
          "key" + i, ("飞哥测试事务消息" + tags[i % tags.length]+"_"+i).getBytes(StandardCharsets.UTF_8));
      TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
      System.out.println("本次发送的消息是=" + new String(message.getBody()));
      System.out.printf("%s%n", transactionSendResult);
      Thread.sleep(10);
    }
    System.out.println("==========所有消息发送完成======");