• 聊聊微服务中消息队列发送消息的4种方式
  • 发布于 1周前
  • 40 热度
    0 评论
微服务开发中经常会使用消息队列进行跨服务通信。在一个典型场景中,服务A执行一个业务逻辑,需要保存数据库,然后通知服务B执行相应的业务逻辑。在这种场景下,我们需要考虑如何发送消息。

一. 基础版
首先,我们可能会考虑将数据库操作和消息发送放在同一个事务中,以下是伪代码示例:
@Transactional  
public void saveWithMessage(BusinessDO businessDO){ 
 String id = IdUtils.nextId();
 businessDO.setId(id);
    xxxRepository.save(businessDO);  
    
    BusinessMessage businessMessage = new BusinessMessage();  
    businessMessage.setKey(id);  
    SendResult send = rocketMQTemplate.syncSend("test-topic", sendMessage);
}
在这段代码里通过@Transactional注解将数据库的操作以及发送消息放到一个事务中,如果数据库的保存或者消息发送失败,则回滚事务。乍一看似乎没什么问题,但稍微推敲一下就会发现此方式有如下两个缺陷:
1.1 数据不一致
首先最容易想到的是,这种消息发送方式无法保证数据的最终一致性。这里先让我来解释一下基于消息队列,生产者发送消息到消费者消费消息的过程:
.生产者发送消息
.MQ收到消息并将数据持久化,在存储中新增一条记录
.返回ACK给生产者
.MQ 推送消息给对应的消费者,等待消费者返回ACK
.如果消费者在指定时间内成功返回ACK,那么MQ则认为消费成功,执行第6步删除消息,如果MQ在指定时间内没有收到ACK,则认为消息消费失败,会重新推送消息,重复执行第4、5、6步操作。
.删除消息。

好,现在回到上面发送消息的场景,假设数据库处理成功,消息消费成功,但是MQ由于某些原因处理超时,导致ACK确认失败,此时整个事务回滚,结果出现数据不一致问题。

这种数据不一致的问题在RPC调用的场景下也经常出现,其根本的原因在于:远程调用,结果最终可能为成功、失败、超时;而对于超时的情况,处理方最终的结果可能是成功,也可能是失败,调用方是无法知晓的。

1.2 事务未提交
其次,使用以上方式还存在另一个问题,即消费者在处理消息时可能读不到刚刚保存的数据,即消费者消费速度快于事务提交的速度。
举例:
假设服务B需要通过消息中的数据ID获取服务A数据库保存的数据。这种情况下,数据库操作与消息发送在同一事务中。可能出现服务B在处理消息时,服务A事务还未提交,导致服务B获取的数据是空数据,无法执行相应业务逻辑。

1.3 适用场景
尽管这种发送方式存在上述两个问题,但在某些场景下仍然适用。例如,消费者在处理时不依赖生产者的数据,且对数据一致性要求不高,这种情况下消息发送和数据库保存可以不在同一个事务中。

二. 进阶版
为解决事务未提交问题,我们可以确保事务提交后再发送消息。在SpringBoot项目中,有两种常见解决方案。
2.1 事务同步器
基于事务同步器的方法:
@Transactional  
public void saveWithMessage(BusinessDO businessDO){  
    xxxRepository.save(businessDO);  
    
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {  
        @Override  
        public void afterCommit() {  
            BusinessMessage businessMessage = new BusinessMessage();  
            businessMessage.setXXX();  
            rocketMQTemplate.syncSend("test-topic", sendMessage);  
        }  
    });  
}
TransactionSynchronizationManager.registerSynchronization 是 Spring 框架中用于注册事务同步的方法。通过这个方法,你可以在事务提交、回滚或完成时执行一些额外的逻辑。在上述代码中,使用了afterCommit方法,在事务成功提交后执行发送消息操作,确保在数据库操作成功且事务稳定的情况下发送消息。

2.2 消息监听器
另一种方法是基于ApplicationEventPublisher,在保存数据库操作后发布一个事件,并通过@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)注解监听事件。这样可以确保消息在数据库事务提交之后再发送。
@Transactional  
public void saveWithMessage(BusinessDO businessDO){  
    xxxRepository.save(businessDO);  

 eventPublisher.publishEvent(new UserCreatedEvent(registerUser));
}

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) 
public void handleUserRegisteredEvent(UserCreatedEvent userCreatedEvent) {  
    rocketMQTemplate.syncSend("test-topic", sendMessage);  
}
这里需要说明一下:在默认情况下Spring的事件监听机制并不是异步的(上次群友弄错了),而是同步的将代码进行解耦,@TransactionalEventListener也是通过同步的方式,但是加入了回调的方式来解决,这样就能够控制事务进行Commited、Rollback时才进行事件的处理,来达到事务同步的目的。

2.3 适用范围
通过以上方式,相较于基础版,可以确保消息在事务提交后发送,解决了消费者读取空数据的问题。但仍然无法保证数据的一致性,适用于对数据一致性要求不高的场景。

三. 本地消息表+补偿重试
如果需要保证最终一致性而非强一致性,可以采用本地消息表+补偿重试的方式来发送消息。
这种方式的执行原理如下:
1.在执行业务操作的同时,在本地消息表中插入一条状态为待发送的记录,业务数据的记录与消息记录必须在同一个事务中完成,这是此方案的核心原则。由于消息表与业务表在同一个库中,事务可以通过数据库来保证。
2.事务提交后发送消息,如果消息发送成功,则将消息状态标记为发送成功或删除消息
3.在生产者服务中会创建一个定时任务,定时从消息表中检索待发送的消息重新发送。
4.对于消费者消费失败,则依赖MQ本身的重试机制来完成,保证数据的最终一致性。

核心代码如下:
@Transactional  
public void saveWithMessage(BusinessDO businessDO){  
    TransactionMessage transactionMessage = new TransactionMessage();  
    transactionMessage.setStaus(MessageStatus.WAITING_SEND);  
    transactionMessage.setMessageKey(businessDO.getId());  
    ...  
  
    xxxRepository.save(businessDO);  
    messageRepository.save(TransactionMessage);  
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {  
        @Override  
        public void afterCommit() {  
            messageService.sendMessage(transactionMessage,businessDO);  
        }  
    });  
}  

public void sendMessage(TransactionMessage transactionMessage,BusinessDO businessDO){  
    BusinessMessage businessMessage = new BusinessMessage();  
    businessMessage.setXXX();  
    try{  
        rocketMQTemplate.syncSend("test-topic", sendMessage);  
        transactionMessage.setStatus(MessageStatus.SUCCESS);  
        messageRepository.update(transactionMessage);  
    }catch (Exception e){  
        // 堆代码 duidaima.com
        // 执行失败的业务逻辑  
    }  
  
}
3.1 问题
虽然这种方式能够保证消息在事务提交后发送,且能够保证最终一致性,但仍然存在一些缺陷:
首先,需要额外的消息表,增加了系统复杂度。(针对此问题,我们又可以将该功能单独提取出来,做成一个消息服务来统一处理,考虑篇幅问题,这里暂不展开。)

其次,通过定时任务轮询消息表,对于处理成功但ACK超时的数据会重新发送消息,这对下游系统产生了强烈的幂等性保障要求,消费者的处理逻辑必须做好幂等控制。关于幂等的处理方案,我在[[Dailymart17:并发与幂等的实现方案]]一文中有详细的说明,欢迎翻阅。

四. 基于事务消息发送
目前,RocketMQ是主流MQ中唯一一个支持事务消息的,如果你们项目恰好使用的是RocketMQ,可以采用事务消息来发送。

小结
本文详细介绍了微服务开发中常用的4种消息发送方式。对于那些对数据一致性要求不高的场景,可以选择使用进阶版的消息发送方式。而对于需要保证最终一致性的情况,推荐采用事务消息和本地消息表的方式进行消息发送。
用户评论