RocketMq 事务消息使用

程序员文章站 2022-07-14 23:04:09

...

在Mq中，由于消息从producer发送出去到最终被消费者消费，中间需要经过mq的服务器进行中转，在rocketMq中即为broker，rabbitmq中为exchange，意思差不多，这样一来消息的投递就有了不确定性，因此在rocketMq中，引入了事务性消息这一概念；

1、RocketMq事务消息是保证事务的最终一致性；

2、半消息，producer消息投递出去了，到达了broker，但是broker未能找到合适的queue，即找到能够消费消息的消费者，这时producer端就无法获得broker对该消息的二次确认，这时候broker会将该消息标记为“暂时不能投递”的状态，即该消息处于半消息状态；

3、消息回查，由于某些特殊原因，如网络抖动，生产者应用重启，可能导致某条消息的二次确认丢失，而broker会不定时的扫描某条长期处于半消息状态的消息，此时会主动向生产者发送询问该条消息的最终状态，(commit或者rollBack)，即对该消息进行回查，这个是broker自动为我们做的，默认时间间隔是1分钟，可通过在broker.conf文件中设置 transactionCheckInterval 的值来改变默认值，单位为毫秒

RocketMq 事务消息使用

上面这张简图描述的是本地的某个事务性方法中，使用rocketMq时候整个消息的传递状态，即如果要保证事务性操作时候，需要配置rocketMq的事务性消息一起使用，下面使用代码来简单模拟一下发送事务性消息的过程，

4、沿用之前使用的整合框架，这里我们新创建一个controller,模拟发送一个事务性消息的接口，

@Controller
public class TransactionController {

	@Autowired
	private TransactionProducer transactionProducer;

	private static final String TOPIC_NAME = "transaction_topic";
	
	//http:/localhost:8082//api/v1/sendTransactionMsg?tag=apk&otherParam=1
	/**
	 * 发送事务性消息
	 * @param tag
	 * @param otherParam
	 * @return
	 * @throws Exception
	 */
	@RequestMapping("/api/v1/sendTransactionMsg")
	@ResponseBody
	public Object sendTransactionMsg(String tag, String otherParam) throws Exception {
		Message message = new Message(TOPIC_NAME, tag, tag + "_key", tag.getBytes());
		SendResult sendResult = transactionProducer.getProducer().sendMessageInTransaction(message, otherParam);
		System.out.printf("发送结果=%s, sendResult=%s \n", sendResult.getSendStatus(), sendResult.toString());
		return "发送成功";
	}

}

5、consumer的代码很简单，直接使用之前的，这里需要重新修改一下消费者的归属组名字，

@Component
public class PayConsumer1 {

	private DefaultMQPushConsumer consumer;

	private String consumerGroup = "transac_consumer_group";

	public PayConsumer1() throws MQClientException {

		consumer = new DefaultMQPushConsumer(consumerGroup);
		consumer.setNamesrvAddr("192.168.111.132:9876");
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		// 默认是集群方式，可以更改为广播，但是广播方式不支持重试
		consumer.setMessageModel(MessageModel.CLUSTERING);
		consumer.subscribe("transaction_topic", "*");
		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				MessageExt msg = msgs.get(0);
				String key = msg.getKeys();
				try {

					System.out.printf("%s 2 Receive New Messages: %s %n", Thread.currentThread().getName(),
							new String(msgs.get(0).getBody()));
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				} catch (Exception e) {
					System.out.println("消费异常");
					e.printStackTrace();
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
			}
		});

		consumer.start();
		System.out.println("consumer start ...");
	}

}

6、接下来是producer类，在之前我们发送消息时候，我们使用的是DefaultMQProducer这个组件，但这里要发送事务性消息的话，需要使用TransactionMQProducer 这个组件，顾名思义可以看出其意思，整个代码如下，

@Component
public class TransactionProducer {

	private String producerGroup = "trac_producer_group";

	//事务监听器
	private TransactionListener transactionListener = new TransactionListenerImpl();
	
	private TransactionMQProducer producer = null;

	// 一般自定义线程池的时候，需要给线程加个名称
	private ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
			new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
				@Override
				public Thread newThread(Runnable r) {
					Thread thread = new Thread(r);
					thread.setName("client-transaction-msg-check-thread");
					return thread;
				}

			});

	
	public TransactionProducer() {
		producer = new TransactionMQProducer(producerGroup);
		producer.setNamesrvAddr("192.168.111.132:9876");
		producer.setTransactionListener(transactionListener);
		producer.setExecutorService(executorService);
		// 指定NameServer地址，多个地址以 ; 隔开

		start();
	}

	public TransactionMQProducer getProducer() {
		return this.producer;
	}

	/**
	 * 对象在使用之前必须要调用一次，只能初始化一次
	 */
	public void start() {
		try {
			this.producer.start();
		} catch (MQClientException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 一般在应用上下文，使用上下文监听器，进行关闭
	 */
	public void shutdown() {
		this.producer.shutdown();
	}

}

/**
 * 实际处理业务的类，可能是本地带事务性的方法中处理
 * @author asus
 */
class TransactionListenerImpl implements TransactionListener{

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        System.out.println("====executeLocalTransaction=======");
        
        String body = new String(msg.getBody());
        String key = msg.getKeys();
        String transactionId = msg.getTransactionId();
        System.out.println("transactionId="+transactionId+", key="+key+", body="+body);
        // 执行本地事务begin TODO
        
        // 执行本地事务end TODO

        int status = Integer.parseInt(arg.toString());

        //二次确认消息，然后消费者可以消费
        if(status == 1){
            return LocalTransactionState.COMMIT_MESSAGE;
        }

        //回滚消息，broker端会删除半消息
        if(status == 2){
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

        //broker端会进行回查消息，再或者什么都不响应
        if(status == 3){
            return LocalTransactionState.UNKNOW;
        }

        return null;
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {

        System.out.println("====checkLocalTransaction=======");
        String body = new String(msg.getBody());
        String key = msg.getKeys();
        String transactionId = msg.getTransactionId();
        System.out.println("transactionId="+transactionId+", key="+key+", body="+body);
        //要么commit 要么rollback
        //可以根据key去检查本地事务消息是否完成
        return LocalTransactionState.COMMIT_MESSAGE;
    }
    
}

在rocketMq的事务性消息中，是通过实现TransactionListener 这个接口实现的，即需要在监听器类里面手动编写代码，在这个里面做处理，这个接口主要有两个方法，executeLocalTransaction ()和checkLocalTransaction()，第一个方法是执行本地事务的，即我们可以在这个方法体里面捕获本地事务的状态然后处理到达broker上的消息，

在目前的rocketMq版本中，提供了这么3种事务性状态，在LocalTransactionState 这个枚举类里面，

COMMIT_MESSAGE,
ROLLBACK_MESSAGE,
UNKNOW,

第一个状态意思是提交消息，即broker确认了这条消息的正确性之后执行提交，标记这条消息可被消费，这样的话consumer就可以正常消费这条消息了；
第二个状态意思是当我们的本地主事务发生异常的时候，回滚本地事务的同时，同样需要一种方法通知到rocketMq不要继续发送消息了，当broker收到这个命令时候就会标记消息为rollBack的状态，consumer就不能收到了；
第三种即为上面描述的消息回查的状态；

然后我们启动一下程序，通过断点的方式调试一下，首先我们模拟一下事务的正常提交，
http://localhost:8082//api/v1/sendTransactionMsg?tag=apk&otherParam=1
RocketMq 事务消息使用

可以看到消息直接进入了监听器中，继续往下执行，可以看到消费者可以正常消费到这条消息，
RocketMq 事务消息使用
下面，我们将参数设置为2，在看一下，消息发送到了broker上，但是被broker回滚了，即这样的消息会被broker删除的，所以无法到达consumer上，
http://localhost:8082//api/v1/sendTransactionMsg?tag=apk&otherParam=2

最后，当我们设置参数为3，即状态为回查的时候，仍然可以看到和状态2同样的结果，但是会进入到checkLocalTransaction 回调方法中，
http://localhost:8082//api/v1/sendTransactionMsg?tag=apk&otherParam=3

基本上到这里，关于rocketMq发送事务性消息的简单使用就结束了，更深入的大家可在此基础上继续研究，有人在模拟消息回查的时候效果没有出来，可能是rocketMq的版本有点儿问题，可以按照我上面说的手动去设置一下参数，把时间间隔调整的短一点，分布式事务结合Mq的使用是比较高级也是比较难理解的部分，比如真实业务中，如果需要使用事务性消息，可能还需要配合业务代码进行重新的规划和设计才能达到比较好的效果，有兴趣的伙伴可以继续深究，最后感谢观看！

上一篇： js中成员变量和局部变量发生奇怪的问题

下一篇：使用@Transactional注解事务，可能出现哪些问题呢

RocketMq 事务消息使用

HTML5之消息通知的使用(Web Notification)

使用Python的Zato发送AMQP消息的教程

详谈：Redis事务和消息订阅

python使用xauth方式登录饭否网然后发消息

使用Python制作自动推送微信消息提醒的备忘录功能

RocketMQ中Producer消息的发送

抖音短视频怎么给别人发私信?抖音发消息功能使用方法介绍

使用Java实现数据库编程—05 事务、视图、索引、备份和恢复

Spring实战之使用XML方式管理声明式事务操作示例

MYSQL存储过程中事务和DECLARE EXIT/CONTINUE HANDLER的使用