RocketMQ（09）——发送事务消息

超、凢脫俗 2023-07-13 04:19 45阅读 0赞

# 发送事务消息 #

RocketMQ支持发送事务消息，它的事务消息是基于二阶段提交机制实现的。当发送的消息是事务消息时，只有对应的消息被提交了才能被消费者进行消费。发送事务消息时生产者需要使用`TransactionMQProducer`，它还需要指定一个`TransactionListener`。TransactionListener接口的定义如下。

public interface TransactionListener { 
        /** * When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction. * * @param msg Half(prepare) message * @param arg Custom business parameter * @return Transaction state */
        LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);
    
        /** * When no response to prepare(half) message. broker will send check message to check the transaction status, and this * method will be invoked to get local transaction status. * * @param msg Check message * @return Transaction state */
        LocalTransactionState checkLocalTransaction(final MessageExt msg);
    }

如你所见，TransactionListener定义了两个接口方法，当消息发送到了Broker后，会回调其`executeLocalTransaction()`，根据返回状态来决定是要提交事务还是回滚事务。LocalTransactionState有三种状态，COMMIT\_MESSAGE、ROLLBACK\_MESSAGE和UNKNOW。前两种状态很明显，表示提交和回滚。当在`executeLocalTransaction()`中还不能确定消息是要提交还是回滚时，即可以返回UNKNOW，表示不清楚。这样Broker会发送check消息过来检测事务的状态，到了生产者这里的表现就是回调TransactionListener的`checkLocalTransaction()`方法。下面是一个使用TransactionMQProducer发送事务消息的示例。

@Test
    public void transactionalSend() throws Exception { 
      TransactionMQProducer producer = new TransactionMQProducer("group1");
      producer.setNamesrvAddr(this.nameServer);
      producer.setTransactionListener(new TransactionListener() { 
        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { 
          Integer attach = (Integer) arg;
          if (attach % 6 == 0) { 
            return LocalTransactionState.COMMIT_MESSAGE;
          } else if (attach % 4 == 0) { 
            return LocalTransactionState.ROLLBACK_MESSAGE;
          }
          return LocalTransactionState.UNKNOW;
        }
    
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) { 
          int i = Integer.parseInt(new String(msg.getBody()));
          System.out.println("checkLocalTransaction ----" + i);
          if (i % 10 == 0) { 
            return LocalTransactionState.COMMIT_MESSAGE;
          } else if (i % 7 == 0) { 
            return LocalTransactionState.UNKNOW;
          }
          return LocalTransactionState.ROLLBACK_MESSAGE;
        }
      });
      producer.start();
      for (int i=0; i<100; i++) { 
        Message message = new Message("topic1", "transactional", String.valueOf(i).getBytes());
        producer.sendMessageInTransaction(message, i);
      }
      TimeUnit.SECONDS.sleep(60);
      producer.shutdown();
    }

如上，在调用TransactionMQProducer的`sendMessageInTransaction()`时还可以传递一个业务参数，该参数将作为TransactionListener的`executeLocalTransaction()`的第二个参数。按照上面的代码只有发送消息的顺序（从0开始算）是6的倍数或者是非4的倍数但是10的倍数的消息才会提交事务，对应的消息才能被消费者进行消费。

（注：本文是基于RocketMQ4.5.0所写）