RocketMQ分布式事务 -> 最终一致性实现

这篇具有很好参考价值的文章主要介绍了RocketMQ分布式事务 -> 最终一致性实现。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

前言

· 分布式事务的问题常在业务与面试中被提及, 近日摸鱼看到这篇文章, 阐述的非常通俗易懂, 固持久化下来我博客中, 也以便于我二刷
转载源: 基于RocketMQ分布式事务 - 完整示例

  • 本文代码不只是简单的demo,考虑到一些异常情况、幂等性消费和死信队列等情况,尽量向可靠业务场景靠拢。

事务消息

在这里,笔者不想使用大量的文字赘述 RocketMQ事务消息的原理,我们只需要搞明白两个概念。

  • Half Message半消息

暂时不能被 Consumer消费的消息。Producer已经把消息发送到 Broker端,但是此消息的状态被标记为不能投递,处于这种状态下的消息称为半消息。事实上,该状态下的消息会被放在一个叫做 RMQ_SYS_TRANS_HALF_TOPIC的主题下

当 Producer端对它二次确认后,也就是 Commit之后,Consumer端才可以消费到;那么如果是Rollback,该消息则会被删除,永远不会被消费到。

  • 事务状态回查

我们想,可能会因为网络原因、应用问题等,导致Producer端一直没有对这个半消息进行确认,那么这时候 Broker服务器会定时扫描这些半消息,主动找Producer端查询该消息的状态。

当然,什么时候去扫描,包含扫描几次,我们都可以配置,在后文我们再细说。

简而言之,RocketMQ事务消息的实现原理就是基于两阶段提交和事务状态回查,来决定消息最终是提交还是回滚的。

在本文,我们的代码就以 订单服务、积分服务 为例。结合上文来看,整体流程如下:

RocketMQ分布式事务 -> 最终一致性实现,rocketmq,分布式,微服务,java

场景代码示例

订单服务

在订单服务中,我们接收前端的请求创建订单,保存相关数据到本地数据库。

事务日志表

在订单服务中,除了有一张订单表之外,还需要一个事务日志表。 它的定义如下:

CREATE TABLE `transaction_log` (
  `id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '事务ID',
  `business` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '业务标识',
  `foreign_key` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '对应业务表中的主键',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

这张表专门作用于事务状态回查。当提交业务数据时,此表也插入一条数据,它们共处一个本地事务中。通过事务ID查询该表,如果返回记录,则证明本地事务已提交;如果未返回记录,则本地事务可能是未知状态或者是回滚状态。

TransactionMQProducer

我们知道,通过 RocketMQ发送消息,需先创建一个消息发送者。值得注意的是,如果发送事务消息,在这里我们的创建的实例必须是 TransactionMQProducer

@Component
public class TransactionProducer {
	
    private String producerGroup = "order_trans_group";
    private TransactionMQProducer producer;
 
    //用于执行本地事务和事务状态回查的监听器
    @Autowired
    OrderTransactionListener orderTransactionListener;
    //执行任务的线程池
    ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(50));
            
    @PostConstruct
    public void init(){
        producer = new TransactionMQProducer(producerGroup);
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setSendMsgTimeout(Integer.MAX_VALUE);
        producer.setExecutorService(executor);
        producer.setTransactionListener(orderTransactionListener);
        this.start();
    }
    private void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    //事务消息发送 
    public TransactionSendResult send(String data, String topic) throws MQClientException {
        Message message = new Message(topic,data.getBytes());
        return this.producer.sendMessageInTransaction(message, null);
    }
}

上面的代码中,主要就是创建事务消息的发送者。在这里,我们重点关注 OrderTransactionListener,它负责执行本地事务和事务状态回查。

OrderTransactionListener

@Component
public class OrderTransactionListener implements TransactionListener {
 
    @Autowired
    OrderService orderService;
 
    @Autowired
    TransactionLogService transactionLogService;
 
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        logger.info("开始执行本地事务....");
        LocalTransactionState state;
        try{
            String body = new String(message.getBody());
            OrderDTO order = JSONObject.parseObject(body, OrderDTO.class);
            orderService.createOrder(order,message.getTransactionId());
            state = LocalTransactionState.COMMIT_MESSAGE;
            logger.info("本地事务已提交。{}",message.getTransactionId());
        }catch (Exception e){
            logger.info("执行本地事务失败。{}",e);
            state = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return state;
    }
 
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        logger.info("开始回查本地事务状态。{}",messageExt.getTransactionId());
        LocalTransactionState state;
        String transactionId = messageExt.getTransactionId();
        if (transactionLogService.get(transactionId)>0){
            state = LocalTransactionState.COMMIT_MESSAGE;
        }else {
            state = LocalTransactionState.UNKNOW;
        }
        logger.info("结束本地事务状态查询:{}",state);
        return state;
    }
}

在通过 producer.sendMessageInTransaction发送事务消息后,如果消息发送成功,就会调用到这里的executeLocalTransaction方法,来执行本地事务。在这里,它会完成订单数据和事务日志的插入。

该方法返回值 LocalTransactionState 代表本地事务状态,它是一个枚举类。

public enum LocalTransactionState {
    //提交事务消息,消费者可以看到此消息
    COMMIT_MESSAGE,
    //回滚事务消息,消费者不会看到此消息
    ROLLBACK_MESSAGE,
    //事务未知状态,需要调用事务状态回查,确定此消息是提交还是回滚
    UNKNOW;
}

那么, checkLocalTransaction 方法就是用于事务状态查询。在这里,我们通过事务ID查询transaction_log这张表,如果可以查询到结果,就提交事务消息;如果没有查询到,就返回未知状态。

注意,这里还涉及到另外一个问题。如果是返回未知状态,RocketMQ Broker服务器会以1分钟的间隔时间不断回查,直至达到事务回查最大检测数,如果超过这个数字还未查询到事务状态,则回滚此消息。

当然,事务回查的频率和最大次数,我们都可以配置。在 Broker 端,可以通过这样来配置它:

brokerConfig.setTransactionCheckInterval(10000); //回查频率10秒一次
brokerConfig.setTransactionCheckMax(3);  //最大检测次数为3

业务实现类

@Service
public class OrderServiceImpl implements OrderService {
    @Autowired
    OrderMapper orderMapper;
    @Autowired
    TransactionLogMapper transactionLogMapper;
    @Autowired
    TransactionProducer producer;
 
    Snowflake snowflake = new Snowflake(1,1);
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    //执行本地事务时调用,将订单数据和事务日志写入本地数据库
    @Transactional
    @Override
    public void createOrder(OrderDTO orderDTO,String transactionId){
 
        //1.创建订单
        Order order = new Order();
        BeanUtils.copyProperties(orderDTO,order);
        orderMapper.createOrder(order);
 
        //2.写入事务日志
        TransactionLog log = new TransactionLog();
        log.setId(transactionId);
        log.setBusiness("order");
        log.setForeignKey(String.valueOf(order.getId()));
        transactionLogMapper.insert(log);
 
        logger.info("订单创建完成。{}",orderDTO);
    }
 
    //前端调用,只用于向RocketMQ发送事务消息
    @Override
    public void createOrder(OrderDTO order) throws MQClientException {
        order.setId(snowflake.nextId());
        order.setOrderNo(snowflake.nextIdStr());
        producer.send(JSON.toJSONString(order),"order");
    }
}

在订单业务服务类中,我们有两个方法。一个用于向RocketMQ发送事务消息,一个用于真正的业务数据落库。

至于为什么这样做,其实有一些原因的,我们后面再说。

调用

@RestController
public class OrderController {
 
    @Autowired
    OrderService orderService;
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @PostMapping("/create_order")
    public void createOrder(@RequestBody OrderDTO order) throws MQClientException {
        logger.info("接收到订单数据:{}",order.getCommodityCode());
        orderService.createOrder(order);
    }
}

总结

目前已经完成了订单服务的业务逻辑。我们总结流程如下:
RocketMQ分布式事务 -> 最终一致性实现,rocketmq,分布式,微服务,java
考虑到异常情况,这里的要点如下:

  • 第一次调用createOrder,发送事务消息。如果发送失败,导致报错,则将异常返回,此时不会涉及到任何数据安全。
  • 如果事务消息发送成功,但在执行本地事务时发生异常,那么订单数据和事务日志都不会被保存,因为它们是一个本地事务中。
  • 如果执行完本地事务,但未能及时的返回本地事务状态或者返回了未知状态。那么,会由Broker定时回查事务状态,然后根据事务日志表,就可以判断订单是否已完成,并写入到数据库。

基于这些要素,我们可以说,已经保证了订单服务和事务消息的一致性。那么,接下来就是积分服务如何正确的消费订单数据并完成相应的业务操作。

积分服务

在积分服务中,主要就是消费订单数据,然后根据订单内容,给相应用户增加积分。

积分记录表

CREATE TABLE `t_points` (
  `id` bigint(16) NOT NULL COMMENT '主键',
  `user_id` bigint(16) NOT NULL COMMENT '用户id',
  `order_no` bigint(16) NOT NULL COMMENT '订单编号',
  `points` int(4) NOT NULL COMMENT '积分',
  `remarks` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '备注',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

在这里,我们重点关注order_no字段,它是实现幂等消费的一种选择。

消费者启动

@Component
public class Consumer {
 
    String consumerGroup = "consumer-group";
    DefaultMQPushConsumer consumer;
 
    @Autowired
    OrderListener orderListener;
    
    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.subscribe("order","*");
        consumer.registerMessageListener(orderListener);
        consumer.start();
    }
}

启动一个消费者比较简单,我们指定要消费的 topic 和监听器就好了。

消费者监听器

@Component
public class OrderListener implements MessageListenerConcurrently {
 
    @Autowired
    PointsService pointsService;
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        logger.info("消费者线程监听到消息。");
        try{
            for (MessageExt message:list) {
                logger.info("开始处理订单数据,准备增加积分....");
                OrderDTO order  = JSONObject.parseObject(message.getBody(), OrderDTO.class);
                pointsService.increasePoints(order);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
            logger.error("处理消费者数据发生异常。{}",e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}

监听到消息之后,调用业务服务类处理即可。处理完成则返回CONSUME_SUCCESS以提交,处理失败则返回RECONSUME_LATER来重试。

增加积分

在这里,主要就是对积分数据入库。但注意,入库之前需要先做判断,来达到幂等性消费。

@Service
public class PointsServiceImpl implements PointsService {
 
    @Autowired
    PointsMapper pointsMapper;
 
    Snowflake snowflake = new Snowflake(1,1);
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Override
    public void increasePoints(OrderDTO order) {
		
        //入库之前先查询,实现幂等
        if (pointsMapper.getByOrderNo(order.getOrderNo())>0){
            logger.info("积分添加完成,订单已处理。{}",order.getOrderNo());
        }else{
            Points points = new Points();
            points.setId(snowflake.nextId());
            points.setUserId(order.getUserId());
            points.setOrderNo(order.getOrderNo());
            Double amount = order.getAmount();
            points.setPoints(amount.intValue()*10);
            points.setRemarks("商品消费共【"+order.getAmount()+"】元,获得积分"+points.getPoints());
            pointsMapper.insert(points);
            logger.info("已为订单号码{}增加积分。",points.getOrderNo());
        }
    }
}

幂等性消费

实现幂等性消费的方式有很多种,具体怎么做,根据自己的情况来看。

比如,在本例中,我们直接将订单号和积分记录绑定在同一个表中,在增加积分之前,就可以先查询此订单是否已处理过。

或者,我们也可以额外创建一张表,来记录订单的处理情况。

再者,也可以将这些信息直接放到redis缓存里,在入库之前先查询缓存。

不管以哪种方式来做,总的思路就是在执行业务前,必须先查询该消息是否被处理过。那么这里就涉及到一个数据主键问题,在这个例子中,我们以订单号为主键,也可以用事务ID作主键,如果是普通消息的话,我们也可以创建唯一的消息ID作为主键。

消费异常

我们知道,当消费者处理失败后会返回 RECONSUME_LATER ,让消息来重试,默认最多重试16次。

那,如果真的由于特殊原因,消息一直不能被正确处理,那怎么办 ?

我们考虑两种方式来解决这个问题。

第一,在代码中设置消息重试次数,如果达到指定次数,就发邮件或者短信通知业务方人工介入处理。

@Component
public class OrderListener implements MessageListenerConcurrently {
 
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        logger.info("消费者线程监听到消息。");
        for (MessageExt message:list) {
            if (!processor(message)){
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
 
    /**
     * 消息处理,第3次处理失败后,发送邮件通知人工介入
     * @param message
     * @return
     */
    private boolean processor(MessageExt message){
        String body = new String(message.getBody());
        try {
            logger.info("消息处理....{}",body);
            int k = 1/0;
            return true;
        }catch (Exception e){
            if(message.getReconsumeTimes()>=3){
                logger.error("消息重试已达最大次数,将通知业务人员排查问题。{}",message.getMsgId());
                sendMail(message);
                return true;
            }
            return false;
        }
    }
}

第二,等待消息重试最大次数后,进入死信队列。

消息重试最大次数默认是16次,我们也可以在消费者端设置这个次数。

consumer.setMaxReconsumeTimes(3);//设置消息重试最大次数

死信队列的主题名称是 %DLQ% + 消费者组名称,比如在订单数据中,我们设置了消费者组名:

String consumerGroup = "order-consumer-group";

那么这个消费者,对应的死信队列主题名称就是%DLQ%order-consumer-group

RocketMQ分布式事务 -> 最终一致性实现,rocketmq,分布式,微服务,java

如上图,我们还需要点击TOPIC配置,来修改里面的 perm 属性,改为 6 即可。RocketMQ分布式事务 -> 最终一致性实现,rocketmq,分布式,微服务,java

最后就可以通过程序代码监听这个主题,来通知人工介入处理或者直接在控制台查看处理了。通过幂等性消费和对死信消息的处理,基本上就能保证消息一定会被处理。文章来源地址https://www.toymoban.com/news/detail-605689.html

到了这里,关于RocketMQ分布式事务 -> 最终一致性实现的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 深入理解高并发下的MySQL与Redis缓存一致性问题(增删改查数据缓存的一致性、Canal、分布式系统CAP定理、BASE理论、强、弱一致性、顺序、线性、因果、最终一致性)

    一些小型项目,或极少有并发的项目,这些策略在无并发情况下,不会有什么问题。 读数据策略:有缓存则读缓存,然后接口返回。没有缓存,查询出数据,载入缓存,然后接口返回。 写数据策略:数据发生了变动,先删除缓存,再更新数据,等下次读取的时候载入缓存,

    2024年03月20日
    浏览(46)
  • 聊聊分布式架构09——分布式中的一致性协议

    目录 01从集中式到分布式 系统特点 集中式特点 分布式特点 事务处理差异 02一致性协议与Paxos算法 2PC(Two-Phase Commit) 阶段一:提交事务请求 阶段二:执行事务提交 优缺点 3PC(Three-Phase Commit) 阶段一:CanCommit 阶段二:PreCommit 阶段三:doCommit 优缺点 Paxos算法 拜占庭将军问题

    2024年02月08日
    浏览(47)
  • 分布式一致性算法Paxos

            Paxos算法是Lamport宗师提出的一种基于消息传递的分布式一致性算法,是目前公认的解决分布式一致性问题最有效的算法之一。Google Chubby的作者Mike Burrows曾经狂妄的说过这个世界上只有一种一致性算法,那就是Paxos,其它的算法都是残次品。         Paxos算法是

    2023年04月16日
    浏览(59)
  • 分布式系统架构设计之分布式数据存储的扩展方式、主从复制以及分布式一致性

    在分布式系统中,数据存储的扩展是为了适应业务的增长和提高系统的性能。分为水平扩展和垂直扩展两种方式,这两种方式在架构设计和应用场景上有着不同的优势和局限性。 水平扩展是通过增加节点或服务器的数量来扩大整个系统的容量和性能。在数据存储领域,水平扩

    2024年02月03日
    浏览(70)
  • 【分布式】一致性哈希和哈希槽

    当我们拥有了多台存储服务器之后,现在有多个key,希望可以将这些个key均匀的缓存到这些服务器上,可以使用哪些方案呢? 1.1 直接哈希取模 这是一种最容易想到的方法,使用取模算法hash(key)% N,对key进行hash运算后取模,N是机器的数量。key进行hash后的结果对3取模,得

    2024年02月03日
    浏览(47)
  • 分布式系统的一致性级别划分及Zookeeper一致性级别分析

    在谈到Zookeeper的一致性是哪种级别的一致性问题,以及CAP原则中的C是哪一种一致性级别时有些疑惑。 下面是大多数文章中提到的一致性级别 一致性(Consistency)是指多副本(Replications)问题中的数据一致性。可以分为强一致性、顺序一致性与弱一致性。 1.1 强一致性(Stric

    2024年04月12日
    浏览(58)
  • 分布式「走进分布式一致性协议」从2PC、3PC、Paxos 到 ZAB

    设计一个分布式系统必定会遇到一个问题—— 因为分区容忍性(partition tolerance)的存在,就必定要求我们需要在系统可用性(availability)和数据一致性(consistency)中做出权衡 。这就是著名的 CAP 一致性(Consistency)是指多副本(Replications)问题中的数据一致性。关于分布式

    2024年02月03日
    浏览(66)
  • 分布式一致性算法——Paxos 和 Raft 算法

    本文隶属于专栏《100个问题搞定大数据理论体系》,该专栏为笔者原创,引用请注明来源,不足和错误之处请在评论区帮忙指出,谢谢! 本专栏目录结构和参考文献请见100个问题搞定大数据理论体系 Paxos和Raft算法都是 分布式一致性算法 ,它们的目的都是 在一个分布式系统

    2024年01月20日
    浏览(63)
  • Zookeeper分布式一致性协议ZAB源码剖析

    ZAB 协议全称:Zookeeper Atomic Broadcast(Zookeeper 原子广播协议)。 Zookeeper 是一个为分布式应用提供高效且可靠的分布式协调服务。在解决分布式一致性方面,Zookeeper 并没有使用 Paxos ,而是采用了 ZAB 协议,ZAB是Paxos算法的一种简化实现。 ZAB 协议定义:ZAB 协议是为分布式协调服

    2024年02月07日
    浏览(51)
  • Elasticsearch分布式一致性原理剖析(一)-节点篇

    “Elasticsearch分布式一致性原理剖析”系列将会对Elasticsearch的分布式一致性原理进行详细的剖析,介绍其实现方式、原理以及其存在的问题等(基于6.2版本)。 ES目前是最流行的分布式搜索引擎系统,其使用Lucene作为单机存储引擎并提供强大的搜索查询能力。学习其搜索原理,则

    2024年01月24日
    浏览(72)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包