1. 案例介绍

1.1 业务分析

模拟电商网站购物场景中的【下单】和【支付】业务

###1）下单

用户请求订单系统下单
订单系统通过RPC调用订单服务下单
订单服务调用优惠券服务，扣减优惠券
订单服务调用调用库存服务，校验并扣减库存
订单服务调用用户服务，扣减用户余额
订单服务完成确认订单

###2）支付

用户请求支付系统
支付系统调用第三方支付平台API进行发起支付流程
用户通过第三方支付平台支付成功后，第三方支付平台回调通知支付系统
支付系统调用订单服务修改订单状态
支付系统调用积分服务添加积分
支付系统调用日志服务记录日志

1.2 问题分析

问题1

用户提交订单后，扣减库存成功、扣减优惠券成功、使用余额成功，但是在确认订单操作失败，需要对库存、库存、余额进行回退。

如何保证数据的完整性？

使用MQ保证在下单失败后系统数据的完整性

###问题2

用户通过第三方支付平台（支付宝、微信）支付成功后，第三方支付平台要通过回调API异步通知商家支付系统用户支付结果，支付系统根据支付结果修改订单状态、记录支付日志和给用户增加积分。

商家支付系统如何保证在收到第三方支付平台的异步通知时，如何快速给第三方支付凭条做出回应？

通过MQ进行数据分发，提高系统处理性能

2. 技术分析

2.1 技术选型

SpringBoot
SpringCloud
SpringCloudAlibaba
RocketMQ
Mysql

2.2 创建父工程 shop-cloud-parent

2.2.1 修改pom文件

java

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    
    <groupId>com.glls.java2301</groupId>
    <artifactId>shop-cloud-parent</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>shop-common</module>
    </modules>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencyManagement>
        <dependencies>
            <!--springcloud alibaba 的 版本-->
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>2.2.5.RELEASE</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

            <!--springcloud 的版本 -->
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Hoxton.SR8</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

        </dependencies>
    </dependencyManagement>
</project>

2.2.2 创建公共依赖 shop-common

添加依赖

java

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

3. 环境搭建

3.1 数据库

1）优惠券表 trade_coupon

Field	Type	Comment
coupon_id	bigint(50) NOT NULL	优惠券ID
coupon_price	decimal(10,2) NULL	优惠券金额
user_id	bigint(50) NULL	用户ID
order_id	bigint(32) NULL	订单ID
is_used	int(1) NULL	是否使用 0未使用 1已使用
used_time	timestamp NULL	使用时间

2）商品表 trade_goods

Field	Type	Comment
goods_id	bigint(50) NOT NULL	主键
goods_name	varchar(255) NULL	商品名称
goods_number	int(11) NULL	商品库存
goods_price	decimal(10,2) NULL	商品价格
goods_desc	varchar(255) NULL	商品描述
add_time	timestamp NULL	添加时间

3）订单表 trade_order

Field	Type	Comment
order_id	bigint(50) NOT NULL	订单ID
user_id	bigint(50) NULL	用户ID
order_status	int(1) NULL	订单状态 0未确认 1已确认 2已取消 3无效 4退款
pay_status	int(1) NULL	支付状态 0未支付 1支付中 2已支付
shipping_status	int(1) NULL	发货状态 0未发货 1已发货 2已退货
address	varchar(255) NULL	收货地址
consignee	varchar(255) NULL	收货人
goods_id	bigint(50) NULL	商品ID
goods_number	int(11) NULL	商品数量
goods_price	decimal(10,2) NULL	商品价格
goods_amount	decimal(10,0) NULL	商品总价
shipping_fee	decimal(10,2) NULL	运费
order_amount	decimal(10,2) NULL	订单价格
coupon_id	bigint(50) NULL	优惠券ID
coupon_paid	decimal(10,2) NULL	优惠券
money_paid	decimal(10,2) NULL	已付金额
pay_amount	decimal(10,2) NULL	支付金额
add_time	timestamp NULL	创建时间
confirm_time	timestamp NULL	订单确认时间
pay_time	timestamp NULL	支付时间

4）订单商品日志表 trade_goods_number_log

Field	Type	Comment
goods_id	int(11) NOT NULL	商品ID
order_id	varchar(32) NOT NULL	订单ID
goods_number	int(11) NULL	库存数量
log_time	datetime NULL	记录时间

5）用户表 trade_user

Field	Type	Comment
user_id	bigint(50) NOT NULL	用户ID
user_name	varchar(255) NULL	用户姓名
user_password	varchar(255) NULL	用户密码
user_mobile	varchar(255) NULL	手机号
user_score	int(11) NULL	积分
user_reg_time	timestamp NULL	注册时间
user_money	decimal(10,0) NULL	用户余额

6）用户余额日志表 trade_user_money_log

Field	Type	Comment
user_id	bigint(50) NOT NULL	用户ID
order_id	bigint(50) NOT NULL	订单ID
money_log_type	int(1) NOT NULL	日志类型 1订单付款 2 订单退款
use_money	decimal(10,2) NULL	操作金额
create_time	timestamp NULL	日志时间

7）订单支付表 trade_pay

Field	Type	Comment
pay_id	bigint(50) NOT NULL	支付编号
order_id	bigint(50) NULL	订单编号
pay_amount	decimal(10,2) NULL	支付金额
is_paid	int(1) NULL	是否已支付 1否 2是

8）MQ消息生产表 trade__mq_producer_temp

Field	Type	Comment
id	varchar(100) NOT NULL	主键
group_name	varchar(100) NULL	生产者组名
msg_topic	varchar(100) NULL	消息主题
msg_tag	varchar(100) NULL	Tag
msg_key	varchar(100) NULL	Key
msg_body	varchar(500) NULL	消息内容
msg_status	int(1) NULL	0:未处理;1:已经处理
create_time	timestamp NOT NULL	记录时间

9）MQ消息消费表 trade_mq_consumer_log

Field	Type	Comment
msg_id	varchar(50) NULL	消息ID
group_name	varchar(100) NOT NULL	消费者组名
msg_tag	varchar(100) NOT NULL	Tag
msg_key	varchar(100) NOT NULL	Key
msg_body	varchar(500) NULL	消息体
consumer_status	int(1) NULL	0:正在处理;1:处理成功;2:处理失败
consumer_times	int(1) NULL	消费次数
consumer_timestamp	timestamp NULL	消费时间
remark	varchar(500) NULL	备注

3.2 项目初始化

shop系统基于Maven进行项目管理

3.1.1 工程浏览

父工程：shop-cloud-parent

优惠券服务：shop-coupon-service

依赖

java

  <dependencies>
        <dependency>
            <groupId>com.glls.java2301</groupId>
            <artifactId>shop-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>com.glls.java2301</groupId>
            <artifactId>shop-pojo</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.2</version>
        </dependency>

        <!--    mysql驱动    -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <!--   druid     -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.1.10</version>
        </dependency>


        <!--    web的场景依赖    -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!--     nacos场景依赖   -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>

        <!--nacos配置中心-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
        </dependency>


        <!--rocketmq整合sb依赖-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>

        </dependency>
    </dependencies>

订单服务：shop-order-service

依赖同优惠卷服务
配置文件application.yml

java

server:
  port: 8083

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: orderProducerGroup

mq:
  order:
    consumer:
      group:
        name: order_orderTopic_cancel_order_group
    topic: orderTopic
    tag:
      confirm: order_confirm
      cancel: order_cancel
  pay:
    consumer:
      group:
        name: pay_payTopic_group
    topic: payTopic
    tag: pay_status

配置文件 bootstrap.yml

java

spring:
  application:
    name: order-service
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848 #指定nacos-server的地址
        namespace: java2301     #  命名空间
        group: shop
      config:
        server-addr: localhost:8848  # 配置中心地址
        namespace: java2301
        group: shop
        file-extension: yml
        extension-configs[0]:
          group: common
          data-id: mybatis-plus.yml
          refresh: true

支付服务：shop-pay-service
- 依赖同优惠卷服务
商品服务：shop-goods-service
- 依赖同优惠卷服务
- 配置文件 application.yml
- 配置文件 bootstrap.yml
用户服务：shop-user-service
- 依赖同优惠卷服务
- 配置文件 application.yml
- 配置文件 bootstrap.yml

实体类：shop-pojo

依赖

java

 <dependencies>
  
        <dependency>
            <groupId>com.glls.java2301</groupId>
            <artifactId>shop-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.2</version>
        </dependency>
  
        <!--mp的复合主键问题-->
        <dependency>
            <groupId>com.github.jeffreyning</groupId>
            <artifactId>mybatisplus-plus</artifactId>
            <version>1.5.1-RELEASE</version>
        </dependency>
  
    </dependencies>

工具工程：shop-common

共7个系统

3.3 MybatisPlus逆向工程使用

1）代码生成

使用IDEA的MP插件逆向工程针对数据表生成代码

###2）代码导入

将实体类导入到shop-pojo工程
在服务层工程中导入对应的Mapper类和对应配置文件

3.4 公共类介绍

ID生成器
IDWorker：Twitter雪花算法
异常处理类
CustomerException：自定义异常类
CastException：异常抛出类
常量类
ShopCode：系统状态类
响应实体类
R：封装响应状态和响应信息
ResultCodeEnum：枚举

4. 下单业务

4.1 下单基本流程

把生成的 OrderService OrderServiceImpl OrderMapper OrderMapper.xml 放入订单服务对应的位置

application.yml

java

server:
  port: 8083

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: orderProducerGroup

mq:
  order:
    consumer:
      group:
        name: order_orderTopic_cancel_order_group
    topic: orderTopic
    tag:
      confirm: order_confirm
      cancel: order_cancel
  pay:
    consumer:
      group:
        name: pay_payTopic_group
    topic: payTopic
    tag: pay_status

bootstrap.yml

java

spring:
  application:
    name: order-service
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848 #指定nacos-server的地址
        namespace: java2301     #  命名空间
        group: shop
      config:
        server-addr: localhost:8848  # 配置中心地址
        namespace: java2301
        group: shop
        file-extension: yml
        extension-configs[0]:
          group: common
          data-id: mybatis-plus.yml
          refresh: true

远程配置

java

spring:
  datasource:
    password: 123456
    username: root
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/trade?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true


mybatis-plus:
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
  global-config:
    db-config:
      id-type: auto
  mapper-locations: classpath:mapper/*.xml

controller层

java

@RestController
@RequestMapping("/order")
public class OrderController {
    @Autowired
    private OrderService orderService;


    @RequestMapping("/test")
    public R testConfirmOrder(){

        Order tradeOrder = new Order();

        //优惠券id
        Long couponId = 345988230098857984L;
        //商品id
        Long goodsId = 345959443973935104L;
        //用户id
        Long userId= 345963634385633280L;
        //设置优惠券id
        tradeOrder.setCouponId(couponId);
        //设置 商品id
        tradeOrder.setGoodsId(goodsId);
        // 设置 用户id
        tradeOrder.setUserId(userId);
        //商品数量
        tradeOrder.setGoodsNumber(1);
        //商品单价
        tradeOrder.setGoodsPrice(new BigDecimal(5000));
        // 运费
        tradeOrder.setShippingFee(BigDecimal.ZERO);
        //订单总金额
        tradeOrder.setOrderAmount(new BigDecimal(5000));

        //已付金额   就是使用的余额  账户上有钱 选择使用账户上的钱
        tradeOrder.setMoneyPaid(new BigDecimal(200));
        // 订单地址
        tradeOrder.setAddress("郑州");


        R r = orderService.confirmOrder(tradeOrder);

        return r;
    }
}

1）接口定义

IOrderService

java

public interface IOrderService {
    /**
     * 确认订单
     * @param order
     * @return Result
     */
    Result confirmOrder(TradeOrder order);
}

###2）业务类实现

java

@Slf4j
@Component
@Service(interfaceClass = IOrderService.class)
public class OrderServiceImpl implements IOrderService {

    @Override
    public Result confirmOrder(TradeOrder order) {
        //1.校验订单
        
        //2.生成预订单
       
        try {
            //3.扣减库存
            
            //4.扣减优惠券
           
            //5.使用余额
           
            //6.确认订单
            
            //7.返回成功状态
           
        } catch (Exception e) {
            //1.确认订单失败,发送消息
            
            //2.返回失败状态
        }

    }
}

###3）校验订单

java

private void checkOrder(Order order) {
        //1.校验订单是否存在
        if(order==null){
            CastException.cast(ShopCode.SHOP_ORDER_INVALID);
        }

        //2.校验订单中的商品是否存在
        // feign 远程调用  根据商品id 查询商品是否存在
        Goods goods = goodsApi.findOne(order.getGoodsId());
        if(goods==null){
            CastException.cast(ShopCode.SHOP_GOODS_NO_EXIST);
        }

        //3.校验下单用户是否存在
        // feign 远程调用  根据用户id 查询用户是否存在
        User user = userApi.findOne(order.getUserId());
        if(user==null){
            CastException.cast(ShopCode.SHOP_USER_NO_EXIST);
        }

        //4.校验商品单价是否合法
        if(order.getGoodsPrice().compareTo(goods.getGoodsPrice())!=0){
            CastException.cast(ShopCode.SHOP_GOODS_PRICE_INVALID);
        }

        //5.校验订单商品数量是否合法
        if(order.getGoodsNumber() > goods.getGoodsNumber()){
            CastException.cast(ShopCode.SHOP_GOODS_NUM_NOT_ENOUGH);
        }

        log.info("校验订单通过");
    }

feign接口

java

@FeignClient("goods-service")
public interface GoodsApi {

    @GetMapping("/goods/{goodsId}")
    Goods findOne(@PathVariable("goodsId") Long goodsId);
}

java

@FeignClient("user-service")
public interface UserApi {

    @GetMapping("/user/{userId}")
    User findOne(@PathVariable("userId") Long userId);
}

###4）生成预订单

java

private Long savePreOrder(TradeOrder order) {
        //1.设置订单状态为不可见
        order.setOrderStatus(ShopCode.SHOP_ORDER_NO_CONFIRM.getCode());
        //2.订单ID
        order.setOrderId(idWorker.nextId());
        //核算运费是否正确
        BigDecimal shippingFee = calculateShippingFee(order.getOrderAmount());
    	
        if (order.getShippingFee().compareTo(shippingFee) != 0) {
            CastException.cast(ShopCode.SHOP_ORDER_SHIPPINGFEE_INVALID);
        }
        //3.计算订单总价格是否正确
        BigDecimal orderAmount = order.getGoodsPrice().multiply(new BigDecimal(order.getGoodsNumber()));
        orderAmount.add(shippingFee);
        if (orderAmount.compareTo(order.getOrderAmount()) != 0) {
            CastException.cast(ShopCode.SHOP_ORDERAMOUNT_INVALID);
        }

        //4.判断优惠券信息是否合法
        Long couponId = order.getCouponId();
        if (couponId != null) {
            TradeCoupon coupon = couponService.findOne(couponId);
            //优惠券不存在
            if (coupon == null) {
                CastException.cast(ShopCode.SHOP_COUPON_NO_EXIST);
            }
            //优惠券已经使用
            if ((ShopCode.SHOP_COUPON_ISUSED.getCode().toString())
                .equals(coupon.getIsUsed().toString())) {
                CastException.cast(ShopCode.SHOP_COUPON_INVALIED);
            }
            order.setCouponPaid(coupon.getCouponPrice());
        } else {
            order.setCouponPaid(BigDecimal.ZERO);
        }

        //5.判断余额是否正确
        BigDecimal moneyPaid = order.getMoneyPaid();
        if (moneyPaid != null) {
            //比较余额是否大于0
            int r = order.getMoneyPaid().compareTo(BigDecimal.ZERO);
            //余额小于0
            if (r == -1) {
                CastException.cast(ShopCode.SHOP_MONEY_PAID_LESS_ZERO);
            }
            //余额大于0
            if (r == 1) {
                //查询用户信息
                TradeUser user = userService.findOne(order.getUserId());
                if (user == null) {
                    CastException.cast(ShopCode.SHOP_USER_NO_EXIST);
                }
            //比较余额是否大于用户账户余额
            if (user.getUserMoney().compareTo(order.getMoneyPaid().longValue()) == -1) {
                CastException.cast(ShopCode.SHOP_MONEY_PAID_INVALID);
            }
            order.setMoneyPaid(order.getMoneyPaid());
        }
    } else {
        order.setMoneyPaid(BigDecimal.ZERO);
    }
    //计算订单支付总价
    order.setPayAmount(orderAmount.subtract(order.getCouponPaid())
                       .subtract(order.getMoneyPaid()));
    //设置订单添加时间
    order.setAddTime(new Date());

    //保存预订单
    int r = orderMapper.insert(order);
    if (ShopCode.SHOP_SUCCESS.getCode() != r) {
        CastException.cast(ShopCode.SHOP_ORDER_SAVE_ERROR);
    }
    log.info("订单:["+order.getOrderId()+"]预订单生成成功");
    return order.getOrderId();
}

###5）扣减库存

通过dubbo调用商品服务完成扣减库存

java

private void reduceGoodsNum(TradeOrder order) {
        TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
        goodsNumberLog.setGoodsId(order.getGoodsId());
        goodsNumberLog.setOrderId(order.getOrderId());
        goodsNumberLog.setGoodsNumber(order.getGoodsNumber());
        Result result = goodsService.reduceGoodsNum(goodsNumberLog);
        if (result.getSuccess().equals(ShopCode.SHOP_FAIL.getSuccess())) {
            CastException.cast(ShopCode.SHOP_REDUCE_GOODS_NUM_FAIL);
        }
        log.info("订单:["+order.getOrderId()+"]扣减库存["+order.getGoodsNumber()+"个]成功");
    }

商品服务GoodsService扣减库存

java

@Override
public Result reduceGoodsNum(TradeGoodsNumberLog goodsNumberLog) {
    if (goodsNumberLog == null ||
            goodsNumberLog.getGoodsNumber() == null ||
            goodsNumberLog.getOrderId() == null ||
            goodsNumberLog.getGoodsNumber() == null ||
            goodsNumberLog.getGoodsNumber().intValue() <= 0) {
        CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
    }
    TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsNumberLog.getGoodsId());
    if(goods.getGoodsNumber()<goodsNumberLog.getGoodsNumber()){
        //库存不足
        CastException.cast(ShopCode.SHOP_GOODS_NUM_NOT_ENOUGH);
    }
    //减库存
    goods.setGoodsNumber(goods.getGoodsNumber()-goodsNumberLog.getGoodsNumber());
    goodsMapper.updateByPrimaryKey(goods);


    //记录库存操作日志
    goodsNumberLog.setGoodsNumber(-(goodsNumberLog.getGoodsNumber()));
    goodsNumberLog.setLogTime(new Date());
    goodsNumberLogMapper.insert(goodsNumberLog);

    return new Result(ShopCode.SHOP_SUCCESS.getSuccess(),ShopCode.SHOP_SUCCESS.getMessage());
}

###6）扣减优惠券

通过dubbo完成扣减优惠券

java

private void changeCoponStatus(TradeOrder order) {
    //判断用户是否使用优惠券
    if (!StringUtils.isEmpty(order.getCouponId())) {
        //封装优惠券对象
        TradeCoupon coupon = couponService.findOne(order.getCouponId());
        coupon.setIsUsed(ShopCode.SHOP_COUPON_ISUSED.getCode());
        coupon.setUsedTime(new Date());
        coupon.setOrderId(order.getOrderId());
        Result result = couponService.changeCouponStatus(coupon);
        //判断执行结果
        if (result.getSuccess().equals(ShopCode.SHOP_FAIL.getSuccess())) {
            //优惠券使用失败
            CastException.cast(ShopCode.SHOP_COUPON_USE_FAIL);
        }
        log.info("订单:["+order.getOrderId()+"]使用扣减优惠券["+coupon.getCouponPrice()+"元]成功");
    }

}

优惠券服务CouponService更改优惠券状态

java

@Override
public Result changeCouponStatus(TradeCoupon coupon) {
    try {
        //判断请求参数是否合法
        if (coupon == null || StringUtils.isEmpty(coupon.getCouponId())) {
            CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
        }
		//更新优惠券状态为已使用
        couponMapper.updateByPrimaryKey(coupon);
        return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
    } catch (Exception e) {
        return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
    }
}

###7）扣减用户余额

通过用户服务完成扣减余额

java

private void reduceMoneyPaid(TradeOrder order) {
    //判断订单中使用的余额是否合法
    if (order.getMoneyPaid() != null && order.getMoneyPaid().compareTo(BigDecimal.ZERO) == 1) {
        TradeUserMoneyLog userMoneyLog = new TradeUserMoneyLog();
        userMoneyLog.setOrderId(order.getOrderId());
        userMoneyLog.setUserId(order.getUserId());
        userMoneyLog.setUseMoney(order.getMoneyPaid());
        userMoneyLog.setMoneyLogType(ShopCode.SHOP_USER_MONEY_PAID.getCode());
        //扣减余额
        Result result = userService.changeUserMoney(userMoneyLog);
        if (result.getSuccess().equals(ShopCode.SHOP_FAIL.getSuccess())) {
            CastException.cast(ShopCode.SHOP_USER_MONEY_REDUCE_FAIL);
        }
        log.info("订单:["+order.getOrderId()+"扣减余额["+order.getMoneyPaid()+"元]成功]");
    }
}

用户服务UserService,更新余额

java

@Override
public Result changeUserMoney(TradeUserMoneyLog userMoneyLog) {
    //判断请求参数是否合法
    if (userMoneyLog == null
            || userMoneyLog.getUserId() == null
            || userMoneyLog.getUseMoney() == null
            || userMoneyLog.getOrderId() == null
            || userMoneyLog.getUseMoney().compareTo(BigDecimal.ZERO) <= 0) {
        CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
    }

    //查询该订单是否存在付款记录
    TradeUserMoneyLogExample userMoneyLogExample = new TradeUserMoneyLogExample();
    userMoneyLogExample.createCriteria()
            .andUserIdEqualTo(userMoneyLog.getUserId())
            .andOrderIdEqualTo(userMoneyLog.getOrderId());
   int count = userMoneyLogMapper.countByExample(userMoneyLogExample);
   TradeUser tradeUser = new TradeUser();
   tradeUser.setUserId(userMoneyLog.getUserId());
   tradeUser.setUserMoney(userMoneyLog.getUseMoney().longValue());
   //判断余额操作行为
   //【付款操作】
   if (userMoneyLog.getMoneyLogType().equals(ShopCode.SHOP_USER_MONEY_PAID.getCode())) {
           //订单已经付款，则抛异常
           if (count > 0) {
                CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY);
            }
       	   //用户账户扣减余额
           userMapper.reduceUserMoney(tradeUser);
       }
    //【退款操作】
    if (userMoneyLog.getMoneyLogType().equals(ShopCode.SHOP_USER_MONEY_REFUND.getCode())) {
         //如果订单未付款,则不能退款,抛异常
         if (count == 0) {
         CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY);
     }
     //防止多次退款
     userMoneyLogExample = new TradeUserMoneyLogExample();
     userMoneyLogExample.createCriteria()
             .andUserIdEqualTo(userMoneyLog.getUserId())
                .andOrderIdEqualTo(userMoneyLog.getOrderId())
                .andMoneyLogTypeEqualTo(ShopCode.SHOP_USER_MONEY_REFUND.getCode());
     count = userMoneyLogMapper.countByExample(userMoneyLogExample);
     if (count > 0) {
         CastException.cast(ShopCode.SHOP_USER_MONEY_REFUND_ALREADY);
     }
     	//用户账户添加余额
        userMapper.addUserMoney(tradeUser);
    }


    //记录用户使用余额日志
    userMoneyLog.setCreateTime(new Date());
    userMoneyLogMapper.insert(userMoneyLog);
    return new Result(ShopCode.SHOP_SUCCESS.getSuccess(),ShopCode.SHOP_SUCCESS.getMessage());
}

###8）确认订单

java

private void updateOrderStatus(TradeOrder order) {
    order.setOrderStatus(ShopCode.SHOP_ORDER_CONFIRM.getCode());
    order.setPayStatus(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY.getCode());
    order.setConfirmTime(new Date());
    int r = orderMapper.updateByPrimaryKey(order);
    if (r <= 0) {
        CastException.cast(ShopCode.SHOP_ORDER_CONFIRM_FAIL);
    }
    log.info("订单:["+order.getOrderId()+"]状态修改成功");
}

9）小结

java

@Override
public Result confirmOrder(TradeOrder order) {
    //1.校验订单
    checkOrder(order);
    //2.生成预订单
    Long orderId = savePreOrder(order);
    order.setOrderId(orderId);
    try {
        //3.扣减库存
        reduceGoodsNum(order);
        //4.扣减优惠券
        changeCoponStatus(order);
        //5.使用余额
        reduceMoneyPaid(order);
        //6.确认订单
        updateOrderStatus(order);
        log.info("订单:["+orderId+"]确认成功");
        return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
    } catch (Exception e) {
        //确认订单失败,发送消息
        ...
        return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
    }
}

4.2 失败补偿机制

4.2.1 消息发送方

配置RocketMQ属性值

properties

#订单服务中  mq 的 相关配置

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: orderProducerGroup

mq:
  order:
    consumer:
      group:
        name: order_orderTopic_cancel_order_group
    topic: orderTopic
    tag:
      confirm: order_confirm
      cancel: order_cancel
  pay:
    consumer:
      group:
        name: pay_payTopic_group
    topic: payTopic
    tag: pay_status

注入模板类和属性值信息

java

 @Autowired
 private RocketMQTemplate rocketMQTemplate;

 @Value("${mq.order.topic}")
 private String topic;

 @Value("${mq.order.tag.cancel}")
 private String cancelTag;

发送下单失败消息

java

@Override
public Result confirmOrder(TradeOrder order) {
    //1.校验订单
    //2.生成预订
    try {
        //3.扣减库存
        //4.扣减优惠券
        //5.使用余额
        //6.确认订单
    } catch (Exception e) {
        //确认订单失败,发送消息
        CancelOrderMQ cancelOrderMQ = new CancelOrderMQ();
        cancelOrderMQ.setOrderId(order.getOrderId());
        cancelOrderMQ.setCouponId(order.getCouponId());
        cancelOrderMQ.setGoodsId(order.getGoodsId());
        cancelOrderMQ.setGoodsNumber(order.getGoodsNumber());
        cancelOrderMQ.setUserId(order.getUserId());
        cancelOrderMQ.setUserMoney(order.getMoneyPaid());
        try {
            sendMessage(topic, 
                        cancelTag, 
                        cancelOrderMQ.getOrderId().toString(), 
                    JSON.toJSONString(cancelOrderMQ));
    } catch (Exception e1) {
        e1.printStackTrace();
            CastException.cast(ShopCode.SHOP_MQ_SEND_MESSAGE_FAIL);
        }
        return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
    }
}

java

private void sendMessage(String topic, String tags, String keys, String body) throws Exception {
    //判断Topic是否为空
    if (StringUtils.isEmpty(topic)) {
        CastException.cast(ShopCode.SHOP_MQ_TOPIC_IS_EMPTY);
    }
    //判断消息内容是否为空
    if (StringUtils.isEmpty(body)) {
        CastException.cast(ShopCode.SHOP_MQ_MESSAGE_BODY_IS_EMPTY);
    }
    //消息体
    Message message = new Message(topic, tags, keys, body.getBytes());
    //发送消息
    rocketMQTemplate.getProducer().send(message);
}

4.2.2 消费接收方

配置RocketMQ属性值

properties

#商品服务 回退库存

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: orderProducerGroup

mq:
  order:
    consumer:
      group:
        name: order_orderTopic_cancel_goods_group
    topic: orderTopic
    tag:
      confirm: order_confirm
      cancel: order_cancel

创建监听类，消费消息

java

@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}", 
                         consumerGroup = "${mq.order.consumer.group.name}",
                         messageModel = MessageModel.BROADCASTING)
public class CancelOrderConsumer implements RocketMQListener<MessageExt>{

    @Override
    public void onMessage(MessageExt messageExt) {
        ...
    }
}

1）回退库存

流程分析

这里为什么要查询消息的消费记录呢？这是因为我们要保证消息处理的幂等性。我们会将处理过的消息存到数据库中的trade_mq_consumer_log 表中，我们拿到一条消息不是拿来立马就进行消费而是先判断这个消息是否被处理过，如果被处理过就无须再次处理，否则就会出现幂等性问题说白了避免消息重复消费。

这个表中的consumer_status 这个字段会记录这个消息的处理结果。

consumer_times 指的是消费次数这里最多不能超过三次可以消费失败消费失败的话就可以重复消息但是消费失败的次数一共不能超过三次超过的话也会舍弃消息

查询消息的消费记录如果查询到了则查看消息的状态如果处理成功则直接结束，如果是正在处理也直接结束，如果处理失败则查询失败次数如果不足三次则使用数据库的乐观锁更改消息的状态避免并发问题

如果查询消息消费记录没有查询到，说明这个消息没有被处理过，没有被处理过则在这个表添加一条记录记录这个消息正在被处理的状态

消息消费者

java

/**

这里的泛型指定使用  MessageExt  这个是Message 的子类  ，所以可以使用这个子类来接受 
使用这个类 就可以使用  他的msgId 属性
*/


@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{


    @Value("${mq.order.consumer.group.name}")
    private String groupName;

    @Autowired
    private TradeGoodsMapper goodsMapper;

    @Autowired
    private TradeMqConsumerLogMapper mqConsumerLogMapper;

    @Autowired
    private TradeGoodsNumberLogMapper goodsNumberLogMapper;

    @Override
    public void onMessage(MessageExt messageExt) {
        String msgId=null;
        String tags=null;
        String keys=null;
        String body=null;
        try {
            //1. 解析消息内容
            msgId = messageExt.getMsgId();
            tags= messageExt.getTags();
            keys= messageExt.getKeys();
            body= new String(messageExt.getBody(),"UTF-8");

            log.info("接受消息成功");

            //2. 查询消息消费记录
            TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
            primaryKey.setMsgTag(tags);
            primaryKey.setMsgKey(keys);
            primaryKey.setGroupName(groupName);
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);

            if(mqConsumerLog!=null){
                //3. 判断如果消费过...
                //3.1 获得消息处理状态
                Integer status = mqConsumerLog.getConsumerStatus();
                //处理过...返回
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue()==status.intValue()){
                    log.info("消息:"+msgId+",已经处理过");
                    return;
                }

                //正在处理...返回
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue()==status.intValue()){
                    log.info("消息:"+msgId+",正在处理");
                    return;
                }

                //处理失败
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue()==status.intValue()){
                    //获得消息处理次数
                    Integer times = mqConsumerLog.getConsumerTimes();
                    if(times>3){
                        log.info("消息:"+msgId+",消息处理超过3次,不能再进行处理了");
                        return;
                    }
                    mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());

                    //使用数据库乐观锁更新
                    TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
                    TradeMqConsumerLogExample.Criteria criteria = example.createCriteria();
                    criteria.andMsgTagEqualTo(mqConsumerLog.getMsgTag());
                    criteria.andMsgKeyEqualTo(mqConsumerLog.getMsgKey());
                    criteria.andGroupNameEqualTo(groupName);
                    criteria.andConsumerTimesEqualTo(mqConsumerLog.getConsumerTimes());
                    int r = mqConsumerLogMapper.updateByExampleSelective(mqConsumerLog, example);
                    if(r<=0){
                        //未修改成功,其他线程并发修改
                        log.info("并发修改,稍后处理");
                    }
                }

            }else{
                //4. 判断如果没有消费过...
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(0);

                //将消息处理信息添加到数据库
                mqConsumerLogMapper.insert(mqConsumerLog);
            }
            //5. 回退库存
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            Long goodsId = mqEntity.getGoodsId();
            TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsId);
            goods.setGoodsNumber(goods.getGoodsNumber()+mqEntity.getGoodsNum());
            goodsMapper.updateByPrimaryKey(goods);

            //记录库存操作日志
            TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
            goodsNumberLog.setOrderId(mqEntity.getOrderId());
            goodsNumberLog.setGoodsId(goodsId);
            goodsNumberLog.setGoodsNumber(mqEntity.getGoodsNum());
            goodsNumberLog.setLogTime(new Date());
            goodsNumberLogMapper.insert(goodsNumberLog);

            //6. 将消息的处理状态改为成功
            mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode());
            mqConsumerLog.setConsumerTimestamp(new Date());
            mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
            log.info("回退库存成功");
        } catch (Exception e) {
            e.printStackTrace();
            TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
            primaryKey.setMsgTag(tags);
            primaryKey.setMsgKey(keys);
            primaryKey.setGroupName(groupName);
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);
            if(mqConsumerLog==null){
                //数据库未有记录
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(1);
                mqConsumerLogMapper.insert(mqConsumerLog);
            }else{
                mqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes()+1);
                mqConsumerLogMapper.updateByPrimaryKeySelective(mqConsumerLog);
            }
        }

    }
}

2）回退优惠券

java

@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{


    @Autowired
    private TradeCouponMapper couponMapper;

    @Override
    public void onMessage(MessageExt message) {

        try {
            //1. 解析消息内容
            String body = new String(message.getBody(), "UTF-8");
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            log.info("接收到消息");
            //2. 查询优惠券信息
            TradeCoupon coupon = couponMapper.selectByPrimaryKey(mqEntity.getCouponId());
            //3.更改优惠券状态
            coupon.setUsedTime(null);
            coupon.setIsUsed(ShopCode.SHOP_COUPON_UNUSED.getCode());
            coupon.setOrderId(null);
            couponMapper.updateByPrimaryKey(coupon);
            log.info("回退优惠券成功");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            log.error("回退优惠券失败");
        }

    }
}

3）回退余额

java

@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{


    @Autowired
    private IUserService userService;

    @Override
    public void onMessage(MessageExt messageExt) {

        try {
            //1.解析消息
            String body = new String(messageExt.getBody(), "UTF-8");
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            log.info("接收到消息");
            if(mqEntity.getUserMoney()!=null && mqEntity.getUserMoney().compareTo(BigDecimal.ZERO)>0){
                //2.调用业务层,进行余额修改
                TradeUserMoneyLog userMoneyLog = new TradeUserMoneyLog();
                userMoneyLog.setUseMoney(mqEntity.getUserMoney());
                userMoneyLog.setMoneyLogType(ShopCode.SHOP_USER_MONEY_REFUND.getCode());
                userMoneyLog.setUserId(mqEntity.getUserId());
                userMoneyLog.setOrderId(mqEntity.getOrderId());
                userService.updateMoneyPaid(userMoneyLog);
                log.info("余额回退成功");
            }
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            log.error("余额回退失败");
        }

    }
}

4）取消订单

java

@Override
    public void onMessage(MessageExt messageExt) {
        String body = new String(messageExt.getBody(), "UTF-8");
        String msgId = messageExt.getMsgId();
        String tags = messageExt.getTags();
        String keys = messageExt.getKeys();
        log.info("CancelOrderProcessor receive message:"+messageExt);
        CancelOrderMQ cancelOrderMQ = JSON.parseObject(body, CancelOrderMQ.class);
        TradeOrder order = orderService.findOne(cancelOrderMQ.getOrderId());
		order.setOrderStatus(ShopCode.SHOP_ORDER_CANCEL.getCode());
        orderService.changeOrderStatus(order);
        log.info("订单:["+order.getOrderId()+"]状态设置为取消");
        return order;
    }

4.3 测试

1）准备测试环境

java

@RunWith(SpringRunner.class)
@SpringBootTest(classes = ShopOrderServiceApplication.class)
public class OrderTest {

    @Autowired
    private IOrderService orderService;
}

###1）准备测试数据

用户数据
商品数据
优惠券数据

###2）测试下单成功流程

java

@Test    
public void add(){
    Long goodsId=XXXL;
    Long userId=XXXL;
    Long couponId=XXXL;

    TradeOrder order = new TradeOrder();
    order.setGoodsId(goodsId);
    order.setUserId(userId);
    order.setGoodsNumber(1);
    order.setAddress("北京");
    order.setGoodsPrice(new BigDecimal("5000"));
    order.setOrderAmount(new BigDecimal("5000"));
    order.setMoneyPaid(new BigDecimal("100"));
    order.setCouponId(couponId);
    order.setShippingFee(new BigDecimal(0));
    orderService.confirmOrder(order);
}

执行完毕后,查看数据库中用户的余额、优惠券数据，及订单的状态数据

###3）测试下单失败流程

代码同上。

执行完毕后，查看用户的余额、优惠券数据是否发生更改，订单的状态是否为取消。

5. 支付业务

5.1 创建支付订单

java

public Result createPayment(TradePay tradePay) {
    //查询订单支付状态
    try {
        TradePayExample payExample = new TradePayExample();
        TradePayExample.Criteria criteria = payExample.createCriteria();
        criteria.andOrderIdEqualTo(tradePay.getOrderId());
        criteria.andIsPaidEqualTo(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
        int count = tradePayMapper.countByExample(payExample);
        if (count > 0) {
            CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY);
        }

        long payId = idWorker.nextId();
        tradePay.setPayId(payId);
        tradePay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY.getCode());
        tradePayMapper.insert(tradePay);
        log.info("创建支付订单成功:" + payId);
    } catch (Exception e) {
        return new Result(ShopCode.SHOP_FAIL.getSuccess(), ShopCode.SHOP_FAIL.getMessage());
    }
    return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
}

完成支付订单创建之后就会请求第三方的支付接口用户在第三方的支付平台支付以后第三方平台就会回调我们的接口告诉我们用户的支付结果

5.2 支付回调

5.2.1 流程分析

5.2.2 代码实现

java

public Result callbackPayment(TradePay tradePay) {

    if (tradePay.getIsPaid().equals(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode())) {
        tradePay = tradePayMapper.selectByPrimaryKey(tradePay.getPayId());
        if (tradePay == null) {
            CastException.cast(ShopCode.SHOP_PAYMENT_NOT_FOUND);
        }
        tradePay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
        int i = tradePayMapper.updateByPrimaryKeySelective(tradePay);
        //更新成功代表支付成功
        if (i == 1) {
            TradeMqProducerTemp mqProducerTemp = new TradeMqProducerTemp();
            mqProducerTemp.setId(String.valueOf(idWorker.nextId()));
            mqProducerTemp.setGroupName("payProducerGroup");
            mqProducerTemp.setMsgKey(String.valueOf(tradePay.getPayId()));
            mqProducerTemp.setMsgTag(topic);
            mqProducerTemp.setMsgBody(JSON.toJSONString(tradePay));
            mqProducerTemp.setCreateTime(new Date());
            mqProducerTempMapper.insert(mqProducerTemp);
            TradePay finalTradePay = tradePay;
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        SendResult sendResult = sendMessage(topic, 
                                                            tag, 
                                                            finalTradePay.getPayId(), 
                                                            JSON.toJSONString(finalTradePay));
                        log.info(JSON.toJSONString(sendResult));
                        if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                            mqProducerTempMapper.deleteByPrimaryKey(mqProducerTemp.getId());
                            System.out.println("删除消息表成功");
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        } else {
            CastException.cast(ShopCode.SHOP_PAYMENT_IS_PAID);
        }
    }
    return new Result(ShopCode.SHOP_SUCCESS.getSuccess(), ShopCode.SHOP_SUCCESS.getMessage());
}

线程池优化消息发送逻辑

创建线程池对象

java

@Bean
public ThreadPoolTaskExecutor getThreadPool() {

    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

    executor.setCorePoolSize(4);

    executor.setMaxPoolSize(8);

    executor.setQueueCapacity(100);

    executor.setKeepAliveSeconds(60);

    executor.setThreadNamePrefix("Pool-A");

    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

    executor.initialize();

    return executor;

}

使用线程池

java

@Autowired
private ThreadPoolTaskExecutor executorService;

executorService.submit(new Runnable() {
    @Override
    public void run() {
        try {
            SendResult sendResult = sendMessage(topic, tag, finalTradePay.getPayId(), JSON.toJSONString(finalTradePay));
            log.info(JSON.toJSONString(sendResult));
            if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                mqProducerTempMapper.deleteByPrimaryKey(mqProducerTemp.getId());
                System.out.println("删除消息表成功");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
});

5.2.3 处理消息

支付成功后，支付服务payService发送MQ消息，订单服务、用户服务、日志服务需要订阅消息进行处理

订单服务修改订单状态为已支付
日志服务记录支付日志
用户服务负责给用户增加积分

以下用订单服务为例说明消息的处理情况

1）配置RocketMQ属性值

properties

mq:
  pay:
    consumer:
      group:
        name: pay_payTopic_group
    topic: payTopic
    tag: pay_status

2）消费消息

在订单服务中，消费消息更改订单状态

java

package com.glls.shop.mq;

import com.alibaba.fastjson.JSON;
import com.glls.constant.ShopCode;
import com.glls.shop.mapper.OrderMapper;
import com.glls.shop.pojo.Order;
import com.glls.shop.pojo.Pay;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @date 2022/11/28
 * @desc 接受支付回调接口发来的 支付结果消息  更改订单的支付窗台
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.pay.topic}",consumerGroup = "${mq.pay.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class PayResultListener implements RocketMQListener<MessageExt> {

    @Autowired
    private OrderMapper orderMapper;

    @Override
    public void onMessage(MessageExt messageExt) {
        //TODO   把 订单的状态修改了

        //消息体 json 串的形式
        String body = new String(messageExt.getBody());
        Pay tradePay = JSON.parseObject(body, Pay.class);

        Order tradeOrder = orderMapper.selectById(tradePay.getOrderId());
        tradeOrder.setPayStatus(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());

       orderMapper.updateById(tradeOrder);

        log.info("把订单的支付状态改为已支付");
    }
}

6. 测试

6.1 下单测试

java

package com.glls.shop.controller;

import com.glls.common.R;
import com.glls.shop.pojo.Order;
import com.glls.shop.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.math.BigDecimal;

/**
* @date 2023/6/17
* @desc
*/
@RestController
@RequestMapping("/order")
public class OrderController {
   @Autowired
   private OrderService orderService;


   @RequestMapping("/test")
   public R testConfirmOrder(){

       Order tradeOrder = new Order();

       //优惠券id
       Long couponId = 345988230098857984L;
       //商品id
       Long goodsId = 345959443973935104L;
       //用户id
       Long userId= 345963634385633280L;
       //设置优惠券id
       tradeOrder.setCouponId(couponId);
       //设置 商品id
       tradeOrder.setGoodsId(goodsId);
       // 设置 用户id
       tradeOrder.setUserId(userId);
       //商品数量
       tradeOrder.setGoodsNumber(1);
       //商品单价
       tradeOrder.setGoodsPrice(new BigDecimal(5000));
       // 运费
       tradeOrder.setShippingFee(BigDecimal.ZERO);
       //订单总金额
       tradeOrder.setOrderAmount(new BigDecimal(5000));

       //已付金额   就是使用的余额  账户上有钱 选择使用账户上的钱
       tradeOrder.setMoneyPaid(new BigDecimal(200));
       // 订单地址
       tradeOrder.setAddress("郑州");


       R r = orderService.confirmOrder(tradeOrder);

       return r;
   }
}

6.3 支付测试

java

package com.glls.shop.controller;

import com.glls.common.R;
import com.glls.shop.pojo.Pay;
import com.glls.shop.service.PayService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @date 2023/6/19
 * @desc
 */
@RestController
@RequestMapping("/pay")
public class PayController {

    @Autowired
    private PayService payService;

    //创建支付订单
    @PostMapping("/createPayment")
    public R createPayment(@RequestBody Pay pay){
        int result = payService.createPayment(pay);
        if(result>0){
            return R.ok();
        }else{
            return R.error();
        }
    }


    //调第三方支付接口 完成支付功能

    /**
     * 模拟 第三方 回调咱们的接口  告诉咱们支付结果
     *
     * */
    @PostMapping("/callbackPayment")
    public R callbackPayment(@RequestBody Pay pay){
        return payService.callbackPayment(pay);
    }

}

1. 案例介绍 ​

1.1 业务分析 ​

1.2 问题分析 ​

问题1 ​

2. 技术分析 ​

2.1 技术选型 ​

2.2 创建父工程 shop-cloud-parent ​

3. 环境搭建 ​

3.1 数据库 ​

1）优惠券表 trade_coupon ​

2）商品表 trade_goods ​

3）订单表 trade_order ​

4）订单商品日志表 trade_goods_number_log ​

5）用户表 trade_user ​

6）用户余额日志表 trade_user_money_log ​

7）订单支付表 trade_pay ​

8）MQ消息生产表 trade__mq_producer_temp ​

9）MQ消息消费表 trade_mq_consumer_log ​

3.2 项目初始化 ​

3.1.1 工程浏览 ​

3.3 MybatisPlus逆向工程使用 ​

1）代码生成 ​

3.4 公共类介绍 ​

4. 下单业务 ​

4.1 下单基本流程 ​

1）接口定义 ​

9）小结 ​

4.2 失败补偿机制 ​

4.2.1 消息发送方 ​

4.2.2 消费接收方 ​

1）回退库存 ​

2）回退优惠券 ​

3）回退余额 ​

4）取消订单 ​

4.3 测试 ​

1）准备测试环境 ​

5. 支付业务 ​

5.1 创建支付订单 ​

5.2 支付回调 ​

5.2.1 流程分析 ​

5.2.2 代码实现 ​

线程池优化消息发送逻辑 ​

5.2.3 ​

处理消息 ​

1）配置RocketMQ属性值 ​

2）消费消息 ​

6. 测试 ​

6.1 下单测试 ​

6.3 支付测试 ​

1. 案例介绍

1.1 业务分析

1.2 问题分析

问题1

2. 技术分析

2.1 技术选型

2.2 创建父工程 shop-cloud-parent

3. 环境搭建

3.1 数据库

1）优惠券表 trade_coupon

2）商品表 trade_goods

3）订单表 trade_order

4）订单商品日志表 trade_goods_number_log

5）用户表 trade_user

6）用户余额日志表 trade_user_money_log

7）订单支付表 trade_pay

8）MQ消息生产表 trade__mq_producer_temp

9）MQ消息消费表 trade_mq_consumer_log

3.2 项目初始化

3.1.1 工程浏览

3.3 MybatisPlus逆向工程使用

1）代码生成

3.4 公共类介绍

4. 下单业务

4.1 下单基本流程

1）接口定义

9）小结

4.2 失败补偿机制

4.2.1 消息发送方

4.2.2 消费接收方

1）回退库存

2）回退优惠券

3）回退余额

4）取消订单

4.3 测试

1）准备测试环境

5. 支付业务

5.1 创建支付订单

5.2 支付回调

5.2.1 流程分析

5.2.2 代码实现

线程池优化消息发送逻辑

5.2.3

处理消息

1）配置RocketMQ属性值

2）消费消息

6. 测试

6.1 下单测试

6.3 支付测试