Springboot+RocketMQ通过事务消息优雅的实现订单支付功能

🕗 发布于 2024-01-09 04:33 spring boot java-rocketmq rocketmq

1. 事务消息

RocketMQ的事务消息，是指发送消息事件和其他事件需要同时成功或同时失败。比如银行转账， A银行的某账户要转一万元到B银行的某账户。A银行发送“B银行账户增加一万元”这个消息，要和“从A银行账户扣除一万元”这个操作同时成功或者同时失败。RocketMQ采用两阶段提交的方式实现事务消息。

1.1 RocketMQ事务消息的原理

半事务消息发送：生产者将半事务消息发送至RocketMQ服务端。
消息持久化及返回Ack确认：RocketMQ服务端接收到半事务消息并持久化成功后，向生产者返回Ack确认消息已经发送成功。此时消息状态为半事务消息。
执行本地事务逻辑：根据发送结果执行本地事务，如果写入失败，此时half消息对业务不可见，本地事务逻辑不执行。
提交二次确认结果：根据本地事务状态执行Commit或者Rollback。RocketMQ 的事务消息分为3种状态，分别是提交状态、回滚状态、未知状态。
TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
TransactionStatus.Unknown: 未知状态，它代表需要检查消息队列来确定状态。
消息回查：(1) 对没有Commit/Rollback的事务消息，从服务端发起一次回查 (2) Producer收到回查消息，检查回查消息对应的本地事务的状态 (3) 根据本地事务状态，重新Commit或者Rollback。第一次回查后仍未获取到事务状态，则之后每隔30s会再次回查，最多重试15次，超过了就会默认丢弃此消息。

1.2 RocketMQ订单支付功能设计

数据库设计

/*
SQLyog Community v13.2.0 (64 bit)
MySQL - 8.0.33 : Database - shop
*********************************************************************
*/

/*!40101 SET NAMES utf8 */;

/*!40101 SET SQL_MODE=''*/;

/*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
/*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
/*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;
CREATE DATABASE /*!32312 IF NOT EXISTS*/`shop` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */;

USE `shop`;

/*Table structure for table `shop_order` */

DROP TABLE IF EXISTS `shop_order`;

CREATE TABLE `shop_order` (
  `id` VARCHAR(50) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin NOT NULL COMMENT '订单id',
  `total_num` INT DEFAULT NULL COMMENT '数量合计',
  `moneys` INT DEFAULT NULL COMMENT '金额合计',
  `pay_type` VARCHAR(1) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL COMMENT '支付类型，1、在线支付、0 货到付款',
  `create_time` DATETIME DEFAULT NULL COMMENT '订单创建时间',
  `update_time` DATETIME DEFAULT NULL COMMENT '订单更新时间',
  `pay_time` DATETIME DEFAULT NULL COMMENT '付款时间',
  `consign_time` DATETIME DEFAULT NULL COMMENT '发货时间',
  `end_time` DATETIME DEFAULT NULL COMMENT '交易完成时间',
  `username` VARCHAR(50) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL COMMENT '用户名称',
  `recipients` VARCHAR(50) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL COMMENT '收货人',
  `recipients_mobile` VARCHAR(12) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL COMMENT '收货人手机',
  `recipients_address` VARCHAR(200) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL COMMENT '收货人地址',
  `weixin_transaction_id` VARCHAR(30) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin DEFAULT NULL COMMENT '交易流水号',
  `order_status` INT DEFAULT NULL COMMENT '订单状态,0:未完成,1:已完成，2：已退货',
  `pay_status` INT DEFAULT NULL COMMENT '支付状态,0:未支付，1：已支付，2：支付失败',
  `is_delete` INT DEFAULT NULL COMMENT '是否删除',
  PRIMARY KEY (`id`),
  KEY `create_time` (`create_time`),
  KEY `status` (`order_status`),
  KEY `payment_type` (`pay_type`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_bin;

/*Data for the table `shop_order` */

/*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
/*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
/*!40014 SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS */;
/*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;

添加RocketMQ依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>

bootstrap.yaml配置

server:
  port: 8085
spring:
  application:
    name: mall-order
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/shop?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC
    username: root
    password: 123456
  cloud:
    nacos:
      config:
        file-extension: yaml
        server-addr: localhost:8848
      discovery:
        #Nacos的注册地址
        server-addr: localhost:8848

rocketmq:
  name-server: localhost:9876
  producer:
    group: test-group-producer

Service层

public interface OrderService extends IService<Order> {
       //添加订单
       void add(Order order);
       //修改订单支付状态
       void pay(String id);
}

@Service
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order>
    implements OrderService{

    @Autowired
    OrderMapper orderMapper;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void add(Order order) {
        order.setCreateTime(new Date());
        orderMapper.insert(order); //这里仅仅生成订单，还有扣减库存等等一系列操作省略
    }
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void pay(String id) {
        //模拟支付完成，修改订单的支付状态
        Order order = orderMapper.selectById(id);
        order.setPayStatus(1);
        order.setPayTime(new Date());
        orderMapper.updateById(order);
    }
}

创建生产者

@RestController
@Slf4j
public class TestController {

    @Autowired
    OrderMapper orderMapper;

    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @RequestMapping("/send")
    public String send(){
        String id = UUID.randomUUID().toString();
        String msg = "订单"+id+"支付成功";
        Order order=new Order();
        order.setId(id);
        order.setCreateTime(new Date());
        order.setMoneys(100);
        order.setUsername("张三");
        Message<String> message = MessageBuilder.withPayload(msg).setHeader("key",id).build();
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("order", message, order);
        String transactionId = result.getTransactionId();
        String status = result.getSendStatus().name();
        log.info("发送消息成功 transactionId={} status={} ",transactionId,status);
        return "success";
    }
}

创建消费者

@Component
@Slf4j
@RocketMQMessageListener(consumerGroup = "test-consumer",topic = "order",messageModel = MessageModel.CLUSTERING)
public class RocketMQListen implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
        System.out.println(body);
    }
}

生产者消息监听器

@Component
@RocketMQTransactionListener
public class TransactionMsgListener implements RocketMQLocalTransactionListener {

    @Autowired
    OrderService orderService;



    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        Order order = (Order) o;
        try {
            //生成订单
            orderService.add(order);
            return RocketMQLocalTransactionState.UNKNOWN;
        }catch (Throwable throwable){
            throwable.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String key = message.getHeaders().get("key").toString();
        System.out.println("回查订单id "+key+" 回查时间"+new Date());
        Order order = orderService.getById(key);
        if(order!=null) {
            long l = new Date().getTime() - order.getCreateTime().getTime();
            long time = l / (1000 * 60);
            //超时1分钟后，就会把未支付的订单进行删除
            if (time > 1) {
                orderService.removeById(key);
                System.out.println("订单" + key + "删除");
                //订单，库存等一系列操作
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            Integer payStatus = order.getPayStatus();
            if (payStatus == 1) {
                return RocketMQLocalTransactionState.COMMIT;
            }
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        else
            return RocketMQLocalTransactionState.ROLLBACK;
    }
}

测试

这里通过生产者发送五个事务消息，生成五个订单，然后两个订单在一分钟内修改支付状态为已支付，超时一分钟未支付就会删除订单回退。运行截图如下：

原文地址：https://blog.csdn.net/qq_43649937/article/details/135461751

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：【Linux Shell】11. 输入/输出重定向
下一篇：OpenHarmony基于HDF简单驱动开发实例

学生公寓管理系统
由于某些特殊原因，需要对学生当前入住的床位进行修改时，使用此功能。新生开学后，宿管管理可以先根据学生所在系查询该专业预定的宿舍和具体的床位入住状态信息，为该学生分配一个空闲状态的床位，并修改当前床位状
阅读更多2024-11-15
Windows上安装RabbitMQ 4.0.3教程
通过上述步骤，你应该已经成功在Windows上安装了RabbitMQ 4.0.3，并启用了管理插件。现在，你可以开始使用RabbitMQ来构建你的分布式系统了。希望这篇教程对你有所帮助，如果有任何问题
阅读更多2024-11-15
【蓝桥等考C++真题】蓝桥杯等级考试C++组第13级L13真题原题(含答案)-成绩排序
C++L13 成绩排序(50 分)
阅读更多2024-11-15
Web入门
HTTP（HyperText Transfer Protocol，超文本传输协议）是一种应用层协议。主要用于在Web浏览器和服务器之间传输数据。基于TCP协议：面向连接，确保了数据传输的可靠性和顺序。
阅读更多2024-11-15
Android 最新的AndroidStudio引入依赖失败如何解决？如:Failed to resolve:xxxx
在引入依赖时报错：Failed to resolve: xxx.xxxx:1.1.0。
阅读更多2024-11-15
力扣113：路径总和II
力扣113：路径总和II。C语言
阅读更多2024-11-15
【springboot使用sqlite数据库】Java后台同时使用mysql、sqlite
根据业务的需要，老版程序使用的数据库是sqlite，版本升级成前后台分离模式，因此需要兼容mysql与sqlite数据库同时使用。
阅读更多2024-11-15
力扣.15 三数之和 three-sum
这道题作为 leetcode 的第 15 道题，看起来似曾相识。大概思路可以有下面几种：暴力解法数组排序+二分Hash 优化双指针。
阅读更多2024-11-15
力扣第 53 题：最大子数组和
题目给定一个整数数组nums，目标是找到和最大的连续子数组（最少包含一个元素），并返回其和。例如，在数组中，和最大的子数组为，其和为6。
阅读更多2024-11-15
linux-缓冲区
调用系统调用接口是需要时间成本的，如果使用缓冲区，存储的字符串到一定长度再调用系统调用，能够有效减少调用系统调用的次数。这个缓冲区是语言设置的，缓冲区满足条件就用系统调用接口write将缓冲区的内容刷
阅读更多2024-11-15

Springboot+RocketMQ通过事务消息优雅的实现订单支付功能

1. 事务消息

1.1 RocketMQ事务消息的原理

1.2 RocketMQ订单支付功能设计

相关文章