RabbitMq实战——外卖派单通过补单系统实现分布式事务

简介: RabbitMq实战——外卖派单通过补单系统实现分布式事务

正文


一、分布式事务


请参考之前的文章


二、思路原理


111.png


当派单系统派单成功之后,订单系统报错,此时将会产生分布式事务的问题,派单数据生成,但此时订单数据异常事务回滚,就发生了分布式事务问题。此时解决分布式事务,生成一个订单的消费者,专门去消费生成订单异常时的一个程序,我们称之为补单系统。


三、代码


订单派单


package com.xiaojie.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.xiaojie.entity.Order;
import com.xiaojie.mapper.OrderMapper;
import com.xiaojie.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.UUID;
/**
 * @author xiaojie
 * @version 1.0
 * @description:
 * @date 2021/10/11 22:09
 */
@Service
@Slf4j
public class OrderServiceImpl implements OrderService, RabbitTemplate.ConfirmCallback {
    //定义交换机
    private static final String XIAOJIE_ORDER_EXCHANGE = "xiaojie_order_exchange";
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Override
    @Transactional
    public String saveOrder() {
        Order order = new Order();
        String orderId = UUID.randomUUID().toString();
        order.setOrderId(orderId);
        order.setOrderName("小谷姐姐麻辣烫");
        order.setPayMoney(35.68);
        order.setStatus(1);//假设订单支付完成
        int result = orderMapper.addOrder(order);
        if (result < 0) {
            return "下单失败";
        }
        //发送派单
        String orderJson = JSONObject.toJSONString(order);
        sendDispatchMsg(orderJson);
        //模拟报错
        int i = 1 / 0;
        return orderId;
    }
    @Async
    public void sendDispatchMsg(String jsonMSg) {
        // 设置生产者消息确认机制
        this.rabbitTemplate.setMandatory(true);
        this.rabbitTemplate.setConfirmCallback(this);
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(jsonMSg);
        //将订单数据发送
        rabbitTemplate.convertAndSend(XIAOJIE_ORDER_EXCHANGE, "", jsonMSg, correlationData);
    }
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        if (ack) {
            log.info(">>>>>>>>消息发送成功:correlationData:{},ack:{},s:{}", correlationData, ack, s);
        } else {
            log.info(">>>>>>>消息发送失败{}", ack);
        }
    }
}


补单系统消费端


package com.xiaojie.consumer;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xiaojie.entity.Order;
import com.xiaojie.mapper.OrderMapper;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
 * @author xiaojie
 * @version 1.0
 * @description: 补单消费者
 * @date 2021/10/11 22:37
 */
@Component
public class OrderConsumer {
    @Autowired
    private OrderMapper orderMapper;
    @RabbitListener(queues = {"xiaojie_order_queue"})
    /**
     * @description: 补单消费者,补偿分布式事务解决框架 数据最终一致性
     * @param:
     * @param: message
     * @param: channel
     * @return: void
     * @author xiaojie
     * @date: 2021/10/11 22:41
     */
    public void compensateOrder(Message message, Channel channel) throws IOException {
        // 1.获取消息
        String msg = new String(message.getBody());
        // 2.获取order对象
        Order orderEntity = JSONObject.parseObject(msg, Order.class);
        //根据订单号查询订单是否存在
        Order dbOrder = orderMapper.getOrder(orderEntity.getOrderId());
        if (dbOrder != null) {
            // 手动ack丢弃消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        //订单没有生成,开始补单
        int result = orderMapper.addOrder(orderEntity);
        if (result > 0) {
            // 手动ack 删除该消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}


派单消费者


package com.xiaojie.consumer;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xiaojie.entity.Dispatch;
import com.xiaojie.mapper.DispatchMapper;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
 * @author xiaojie
 * @version 1.0
 * @description: 派单消费者
 * @date 2021/10/11 22:58
 */
@Component
public class DispatchConsumer {
    @Autowired
    private DispatchMapper dispatchMapper;
    @RabbitListener(queues = "dispatch_order_queue")
    public void dispatchConsumer(Message message, Channel channel) throws IOException {
        // 1.获取消息
        String msg = new String(message.getBody());
        // 2.转换json
        JSONObject jsonObject = JSONObject.parseObject(msg);
        String orderId = jsonObject.getString("orderId");
        // 计算分配的快递员id
        Dispatch dispatch=new Dispatch();
        dispatch.setOrderId(orderId);
        //经过一系列的算法得到送餐时间为30分钟
        dispatch.setSendTime(30*60L);
        dispatch.setRiderId(1000012L);
        dispatch.setUserId(15672L);
        // 3.插入我们的数据库
        int result = dispatchMapper.saveDispatch(dispatch);
        if (result > 0) {
            // 手动ack 删除该消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}


完整代码参考:项目中mq-transaction子模块

spring-boot: Springboot整合redis、消息中间件等相关代码

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
4月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
254 2
|
28天前
|
存储 运维 负载均衡
构建高可用性GraphRAG系统:分布式部署与容错机制
【10月更文挑战第28天】作为一名数据科学家和系统架构师,我在构建和维护大规模分布式系统方面有着丰富的经验。最近,我负责了一个基于GraphRAG(Graph Retrieval-Augmented Generation)模型的项目,该模型用于构建一个高可用性的问答系统。在这个过程中,我深刻体会到分布式部署和容错机制的重要性。本文将详细介绍如何在生产环境中构建一个高可用性的GraphRAG系统,包括分布式部署方案、负载均衡、故障检测与恢复机制等方面的内容。
86 4
构建高可用性GraphRAG系统:分布式部署与容错机制
|
2月前
|
消息中间件 网络协议 C#
C#使用Socket实现分布式事件总线,不依赖第三方MQ
`CodeWF.EventBus.Socket` 是一个轻量级的、基于Socket的分布式事件总线系统,旨在简化分布式架构中的事件通信。它允许进程之间通过发布/订阅模式进行通信,无需依赖外部消息队列服务。
C#使用Socket实现分布式事件总线,不依赖第三方MQ
|
2月前
|
NoSQL Java Redis
开发实战:使用Redisson实现分布式延时消息,订单30分钟关闭的另外一种实现!
本文详细介绍了 Redisson 延迟队列(DelayedQueue)的实现原理,包括基本使用、内部数据结构、基本流程、发送和获取延时消息以及初始化延时队列等内容。文章通过代码示例和流程图,逐步解析了延迟消息的发送、接收及处理机制,帮助读者深入了解 Redisson 延迟队列的工作原理。
|
2月前
|
消息中间件 中间件 数据库
NServiceBus:打造企业级服务总线的利器——深度解析这一面向消息中间件如何革新分布式应用开发与提升系统可靠性
【10月更文挑战第9天】NServiceBus 是一个面向消息的中间件,专为构建分布式应用程序设计,特别适用于企业级服务总线(ESB)。它通过消息队列实现服务间的解耦,提高系统的可扩展性和容错性。在 .NET 生态中,NServiceBus 提供了强大的功能,支持多种传输方式如 RabbitMQ 和 Azure Service Bus。通过异步消息传递模式,各组件可以独立运作,即使某部分出现故障也不会影响整体系统。 示例代码展示了如何使用 NServiceBus 发送和接收消息,简化了系统的设计和维护。
55 3
|
2月前
|
消息中间件 存储 监控
消息队列系统中的确认机制在分布式系统中如何实现
消息队列系统中的确认机制在分布式系统中如何实现
|
2月前
|
消息中间件 存储 监控
【10月更文挑战第2天】消息队列系统中的确认机制在分布式系统中如何实现
【10月更文挑战第2天】消息队列系统中的确认机制在分布式系统中如何实现
|
2月前
|
存储 开发框架 .NET
C#语言如何搭建分布式文件存储系统
C#语言如何搭建分布式文件存储系统
75 2
|
2月前
|
消息中间件 存储 监控
消息队列系统中的确认机制在分布式系统中如何实现?
消息队列系统中的确认机制在分布式系统中如何实现?
|
3月前
|
存储 块存储
ceph分布式存储系统常见术语篇
关于Ceph分布式存储系统的常见术语解释和概述。
124 1
ceph分布式存储系统常见术语篇