Springboot+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 */;

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

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

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;
    }
}