RocketMQ学习（二）

文章目录

1. 案例介绍
- 1.1 业务分析
- - 1）下单
  - 2）支付
- 1.2 问题分析
- - 问题1
  - - 如何保证数据的完整性？
    - 使用MQ保证在下单失败后系统数据的完整性
  - 问题2
  - - 如何处理第三方支付平台的异步通知
    - 通过MQ进行数据分发，提高系统处理性能
2. 技术分析
- 2.1 技术选型
- 2.2 SpringBoot整合RocketMQ
- - 2.2.1 消息生产者
  - - 1）添加依赖
    - 2）配置文件
    - 3）启动类
    - 4）测试类
  - 2.2.2 消息消费者
  - - 1）添加依赖
    - 2）配置文件
    - 3）启动类
    - 4）消息监听器
- 2.3 SpringBoot整合Dubbo
- - 2.3.1 搭建Zookeeper集群
  - - 1）准备工作
    - 2）配置集群
    - 3）启动集群
    - zookeeper安装启动配置示例
  - 2.3.2 RPC服务接口
  - 2.3.3 服务提供者
  - - 1）添加依赖
    - 2）配置文件
    - 3）启动类
    - 4）服务实现
  - dubbo-admin管理平台搭建
  - - 示例
  - 2.3.4 服务消费者
  - - 1）添加依赖
    - 2）配置文件
    - 3）启动类
    - 4）Controller
3. 环境搭建
- 3.1 数据库
- - 1）优惠券表
  - 2）商品表
  - 3）订单表
  - 4）订单商品日志表
  - 5）用户表
  - 6）用户余额日志表
  - 7）订单支付表
  - 8）MQ消息生产表
  - shop.sql
- 3.2 项目初始化
- - 3.1.1 工程浏览
  - 3.1.2 工程关系
- 3.3 Mybatis逆向工程使用
- - 1）代码生成
  - 2）代码导入
- 3.4 公共类介绍
4. 下单业务
- ==流程简介*==
- ==下单业务流程图==
- 4.1 下单基本流程
- - 1）接口定义
  - 2）业务类实现
  - 3）校验订单
  - 4）生成预订单
  - 9）小结
- 4.2 失败补偿机制
- - 4.2.1 ==消息发送方==
  - 4.2.2 消费接收方
  - - 1）==回退库存*==
    - 2）回退优惠券
    - 3）==回退余额==
    - 4）取消订单
- 4.3 测试
- - 1）准备测试环境
  - 2）准备测试数据
  - 3）测试下单成功流程
  - 4）测试下单失败流程
5. 支付业务
- ==流程简介*==
- 5.1 创建支付订单
- 5.2 支付回调
- - 5.2.1 流程分析
  - 5.2.2 代码实现
  - - 线程池优化消息发送逻辑
  - 5.2.3 处理消息
  - - 1）配置RocketMQ属性值
    - 2）消费消息
6. 整体联调
- 6.1 准备工作
- - 1）配置RestTemplate类
  - 2）配置请求地址
- 6.2 下单测试
- 6.3 支付测试

1. 案例介绍

1.1 业务分析

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

1）下单

在这里插入图片描述

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

2）支付

在这里插入图片描述

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

1.2 问题分析

问题1

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

如何保证数据的完整性？

在这里插入图片描述

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

在这里插入图片描述

问题2

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

如何处理第三方支付平台的异步通知

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

在这里插入图片描述

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

在这里插入图片描述

2. 技术分析

2.1 技术选型

SpringBoot
Dubbo
Zookeeper
RocketMQ
Mysql

在这里插入图片描述

2.2 SpringBoot整合RocketMQ

下载rocketmq-spring项目

将rocketmq-spring安装到本地仓库（不用安装，mavenrepository中央仓库中有）

mvn install -Dmaven.skip.test=true

2.2.1 消息生产者

1）添加依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.1.RELEASE</version>
</parent>

<properties>
    <rocketmq-spring-boot-starter-version>2.0.3</rocketmq-spring-boot-starter-version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>${rocketmq-spring-boot-starter-version}</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.6</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

</dependencies>

2）配置文件

# application.properties
rocketmq.name-server=192.168.134.3:9876;192.168.134.4:9876
rocketmq.producer.group=my-group

3）启动类

@SpringBootApplication
public class MQProducerApplication {
    public static void main(String[] args) {
        SpringApplication.run(MQSpringBootApplication.class);
    }
}

4）测试类

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {MQSpringBootApplication.class})
public class ProducerTest {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Test
    public void test1(){
        rocketMQTemplate.convertAndSend("zzhua-springboot-rocketmq",
                                        "hello springboot rocketmq");
    }
}

编写好生产者后，运行测试方法即可。

2.2.2 消息消费者

1）添加依赖

同消息生产者

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.1.RELEASE</version>
</parent>

<properties>
    <rocketmq-spring-boot-starter-version>2.0.3</rocketmq-spring-boot-starter-version>
</properties>

<dependencies>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>${rocketmq-spring-boot-starter-version}</version>
    </dependency>
    
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.6</version>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

</dependencies>

2）配置文件

同消息生产者

# application.properties
rocketmq.name-server=192.168.134.3:9876;192.168.134.4:9876
rocketmq.consumer.group=zzhua-group

3）启动类

@SpringBootApplication
public class MQConsumerApplication {
    
    public static void main(String[] args) {
        
        SpringApplication.run(MQSpringBootApplication.class);
    }
    
}

4）消息监听器

@Slf4j
@Component
@RocketMQMessageListener(topic = "zzhua-springboot-rocketmq",
                         consumeMode = ConsumeMode.CONCURRENTLY,
                         consumerGroup = "${rocketmq.consumer.group}")
public class Consumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        log.info("Receive message：" + message);
    }
}

启动消费者应用，接收到了生产者发送的消息。

2.3 SpringBoot整合Dubbo

下载dubbo-spring-boot-starter依赖包

将dubbo-spring-boot-starter安装到本地仓库

mvn install -Dmaven.skip.test=true

在这里插入图片描述

2.3.1 搭建Zookeeper集群

1）准备工作

安装JDK
将Zookeeper上传到服务器
解压Zookeeper，并创建data目录，将conf下的zoo_sample.cfg文件改名为zoo.cfg
建立/user/local/zookeeper-cluster,将解压后的Zookeeper复制到以下三个目录

/usr/local/zookeeper-cluster/zookeeper-1
/usr/local/zookeeper-cluster/zookeeper-2
/usr/local/zookeeper-cluster/zookeeper-3

配置每一个 Zookeeper 的 dataDir（zoo.cfg） clientPort 分别为 2181 2182 2183

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

clientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

clientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

clientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data

2）配置集群

在每个 zookeeper 的 data 目录下创建一个 myid 文件，内容分别是 1、2、3 。这个文件就是记录每个服务器的 ID
在每一个 zookeeper 的 zoo.cfg 配置客户端访问端口（clientPort）和集群服务器 IP 列表。

集群服务器 IP 列表如下

server.1=192.168.25.140:2881:3881
server.2=192.168.25.140:2882:3882
server.3=192.168.25.140:2883:3883

解释：server.服务器ID=服务器IP地址：服务器之间通信端口：服务器之间投票选举端口

3）启动集群

启动集群就是分别启动每个实例。

在这里插入图片描述

检查zookeeper启动状态

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status

在这里插入图片描述

zookeeper安装启动配置示例

上传zookeeper压缩包到服务器，并解压到/usr/local/zookeeper-cluster

在这里插入图片描述

在/usr/local/zookeeper-cluster/zookeeper-3.4.6创建data文件夹，在文件夹中创建myid文件，并且使用vim写入1

在这里插入图片描述

在/usr/local/zookeeper-cluster/zookeeper-3.4.6/conf/ 下的zoo_sample.cfg拷贝一份，名为zoo.cfg文件

在这里插入图片描述

修改zoo.cfg配置，注意dataDir下的目录中的zookeeper-1与clientPort的2181

在这里插入图片描述

将修改好的zookeeper-3.4.6复制3份，名为zookeeper-1，zookeeper-2，zookeeper-3，需要只需要修改zookeeper-2与zookeeper-3这2个文件夹下的data/myid文件和conf/zoo.cfg配置文件中的端口和数据目录位置

在这里插入图片描述

修改zookeeper-2下的data/myid文件和conf/zoo.cfg配置文件中的端口和数据目录位置，同理修改zookeeper-3文件夹下的data/myid文件和conf/zoo.cfg配置文件

在这里插入图片描述

在bin所在目录启动zk，会默认在运行目录下产生zookeeper.out日志文件。在全部启动之后，查看zk的状态

在这里插入图片描述

2.3.2 RPC服务接口

public interface IUserService {
    public String sayHello(String name);
}

2.3.3 服务提供者

1）添加依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.1.RELEASE</version>
</parent>

<dependencies>
    
    <!--dubbo-->
    <dependency>
        <groupId>com.alibaba.spring.boot</groupId>
        <artifactId>dubbo-spring-boot-starter</artifactId>
        <version>2.0.0</version>
    </dependency>
    
	<!--spring-boot-stater-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
        <exclusions>
            <exclusion>
                <artifactId>log4j-to-slf4j</artifactId>
                <groupId>org.apache.logging.log4j</groupId>
            </exclusion>
        </exclusions>
    </dependency>
    
	<!--zookeeper-->
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.4.10</version>
        <exclusions>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
            </exclusion>
            <exclusion>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

    <dependency>
        <groupId>com.101tec</groupId>
        <artifactId>zkclient</artifactId>
        <version>0.9</version>
        <exclusions>
            <exclusion>
                <artifactId>slf4j-log4j12</artifactId>
                <groupId>org.slf4j</groupId>
            </exclusion>
        </exclusions>
    </dependency>
    
	<!--API-->
    <dependency>
        <groupId>com.itheima.demo</groupId>
        <artifactId>dubbo-api</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>

</dependencies>

2）配置文件

# application.properties
spring.application.name=dubbo-demo-provider
spring.dubbo.application.id=dubbo-demo-provider
spring.dubbo.application.name=dubbo-demo-provider
spring.dubbo.registry.address=zookeeper://192.168.25.140:2181;zookeeper://192.168.25.140:2182;zookeeper://192.168.25.140:2183
spring.dubbo.server=true
spring.dubbo.protocol.name=dubbo
 # 服务调用端口
spring.dubbo.protocol.port=20880

3）启动类

@EnableDubboConfiguration
@SpringBootApplication
public class ProviderBootstrap {

    public static void main(String[] args) throws IOException {
        SpringApplication.run(ProviderBootstrap.class,args);
    }

}

4）服务实现

@Component
@Service(interfaceClass = IUserService.class) // dubbo的@Service注解
public class UserServiceImpl implements IUserService{
    @Override
    public String sayHello(String name) {
        return "hello:"+name;
    }
}

dubbo-admin管理平台搭建

将dubbo-admin.war上传到与zookeeper同一服务器，复制粘贴到tomcat的webapps下，启动tomcat，然后访问：服务器ip:8080/dubbo-admin

示例

将dubbo-admin.war复制到tomcat的webapps目录下，如果zookeeper与dubbo-admin不在同一服务器下，需要修改dubbo-admin.war中的WEB-INF/dubbo.properties文件的zookeeper的ip地址，然后启动tomcat即可如下访问dubbo-admin，可以看到服务已经注册上来了。

在这里插入图片描述

2.3.4 服务消费者

1）添加依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.1.RELEASE</version>
</parent>

<dependencies>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!--dubbo-->
    <dependency>
        <groupId>com.alibaba.spring.boot</groupId>
        <artifactId>dubbo-spring-boot-starter</artifactId>
        <version>2.0.0</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
        <exclusions>
            <exclusion>
                <artifactId>log4j-to-slf4j</artifactId>
                <groupId>org.apache.logging.log4j</groupId>
            </exclusion>
        </exclusions>
    </dependency>

    <!--zookeeper-->
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.4.10</version>
        <exclusions>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
            </exclusion>
            <exclusion>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

    <dependency>
        <groupId>com.101tec</groupId>
        <artifactId>zkclient</artifactId>
        <version>0.9</version>
        <exclusions>
            <exclusion>
                <artifactId>slf4j-log4j12</artifactId>
                <groupId>org.slf4j</groupId>
            </exclusion>
        </exclusions>
    </dependency>

    <!--API-->
    <dependency>
        <groupId>com.itheima.demo</groupId>
        <artifactId>dubbo-api</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>

</dependencies>

2）配置文件

# application.properties
spring.application.name=dubbo-demo-consumer
spring.dubbo.application.name=dubbo-demo-consumer
spring.dubbo.application.id=dubbo-demo-consumer
    spring.dubbo.registry.address=zookeeper://192.168.25.140:2181;zookeeper://192.168.25.140:2182;zookeeper://192.168.25.140:2183

3）启动类

@EnableDubboConfiguration
@SpringBootApplication
public class ConsumerBootstrap {
    public static void main(String[] args) {
        SpringApplication.run(ConsumerBootstrap.class);
    }
}

4）Controller

@RestController
@RequestMapping("/user")
public class UserController {

    @Reference // 使用dubbo的@Reference注解
    private IUserService userService;

    @RequestMapping("/sayHello")
    public String sayHello(String name){
        return userService.sayHello(name);
    }

}

可以看到消费者也注册上来了

在这里插入图片描述

3. 环境搭建

3.1 数据库

1）优惠券表

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）商品表

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）订单表

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）订单商品日志表

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）用户表

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）用户余额日志表

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）订单支付表

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消息生产表

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消息消费表

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	备注

shop.sql

/*
SQLyog Ultimate v8.32 
MySQL - 5.5.49 : Database - trade
*********************************************************************
*/


/*!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*/`trade` /*!40100 DEFAULT CHARACTER SET utf8 */;

USE `trade`;

/*Table structure for table `trade_coupon` */

DROP TABLE IF EXISTS `trade_coupon`;

CREATE TABLE `trade_coupon` (
  `coupon_id` bigint(50) NOT NULL COMMENT '优惠券ID',
  `coupon_price` decimal(10,2) DEFAULT NULL COMMENT '优惠券金额',
  `user_id` bigint(50) DEFAULT NULL COMMENT '用户ID',
  `order_id` bigint(32) DEFAULT NULL COMMENT '订单ID',
  `is_used` int(1) DEFAULT NULL COMMENT '是否使用 0未使用 1已使用',
  `used_time` timestamp NULL DEFAULT NULL COMMENT '使用时间',
  PRIMARY KEY (`coupon_id`),
  KEY `FK_trade_coupon` (`user_id`),
  KEY `FK_trade_coupon2` (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_coupon` */

/*Table structure for table `trade_goods` */

DROP TABLE IF EXISTS `trade_goods`;

CREATE TABLE `trade_goods` (
  `goods_id` bigint(50) NOT NULL AUTO_INCREMENT,
  `goods_name` varchar(255) DEFAULT NULL COMMENT '商品名称',
  `goods_number` int(11) DEFAULT NULL COMMENT '商品库存',
  `goods_price` decimal(10,2) DEFAULT NULL COMMENT '商品价格',
  `goods_desc` varchar(255) DEFAULT NULL COMMENT '商品描述',
  `add_time` timestamp NULL DEFAULT NULL COMMENT '添加时间',
  PRIMARY KEY (`goods_id`)
) ENGINE=InnoDB AUTO_INCREMENT=345959443973935105 DEFAULT CHARSET=utf8;

/*Data for the table `trade_goods` */

insert  into `trade_goods`(`goods_id`,`goods_name`,`goods_number`,`goods_price`,`goods_desc`,`add_time`) values (345959443973935104,'华为P30',999,'5000.00','夜间拍照更美','2019-07-09 20:38:00');

/*Table structure for table `trade_goods_number_log` */

DROP TABLE IF EXISTS `trade_goods_number_log`;

CREATE TABLE `trade_goods_number_log` (
  `goods_id` bigint(50) NOT NULL COMMENT '商品ID',
  `order_id` bigint(50) NOT NULL COMMENT '订单ID',
  `goods_number` int(11) DEFAULT NULL COMMENT '库存数量',
  `log_time` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`goods_id`,`order_id`),
  KEY `FK_trade_goods_number_log2` (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_goods_number_log` */

/*Table structure for table `trade_mq_consumer_log` */

DROP TABLE IF EXISTS `trade_mq_consumer_log`;

CREATE TABLE `trade_mq_consumer_log` (
  `msg_id` varchar(50) DEFAULT NULL,
  `group_name` varchar(100) NOT NULL,
  `msg_tag` varchar(100) NOT NULL,
  `msg_key` varchar(100) NOT NULL,
  `msg_body` varchar(500) DEFAULT NULL,
  `consumer_status` int(1) DEFAULT NULL COMMENT '0:正在处理;1:处理成功;2:处理失败',
  `consumer_times` int(1) DEFAULT NULL,
  `consumer_timestamp` timestamp NULL DEFAULT NULL,
  `remark` varchar(500) DEFAULT NULL,
  PRIMARY KEY (`group_name`,`msg_tag`,`msg_key`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_mq_consumer_log` */

/*Table structure for table `trade_mq_producer_temp` */

DROP TABLE IF EXISTS `trade_mq_producer_temp`;

CREATE TABLE `trade_mq_producer_temp` (
  `id` varchar(100) NOT NULL,
  `group_name` varchar(100) DEFAULT NULL,
  `msg_topic` varchar(100) DEFAULT NULL,
  `msg_tag` varchar(100) DEFAULT NULL,
  `msg_key` varchar(100) DEFAULT NULL,
  `msg_body` varchar(500) DEFAULT NULL,
  `msg_status` int(1) DEFAULT NULL COMMENT '0:未处理;1:已经处理',
  `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_mq_producer_temp` */

/*Table structure for table `trade_order` */

DROP TABLE IF EXISTS `trade_order`;

CREATE TABLE `trade_order` (
  `order_id` bigint(50) NOT NULL COMMENT '订单ID',
  `user_id` bigint(50) DEFAULT NULL COMMENT '用户ID',
  `order_status` int(1) DEFAULT NULL COMMENT '订单状态 0未确认 1已确认 2已取消 3无效 4退款',
  `pay_status` int(1) DEFAULT NULL COMMENT '支付状态 0未支付 1支付中 2已支付',
  `shipping_status` int(1) DEFAULT NULL COMMENT '发货状态 0未发货 1已发货 2已收货',
  `address` varchar(255) DEFAULT NULL COMMENT '收货地址',
  `consignee` varchar(255) DEFAULT NULL COMMENT '收货人',
  `goods_id` bigint(50) DEFAULT NULL COMMENT '商品ID',
  `goods_number` int(11) DEFAULT NULL COMMENT '商品数量',
  `goods_price` decimal(10,2) DEFAULT NULL COMMENT '商品价格',
  `goods_amount` decimal(10,0) DEFAULT NULL COMMENT '商品总价',
  `shipping_fee` decimal(10,2) DEFAULT NULL COMMENT '运费',
  `order_amount` decimal(10,2) DEFAULT NULL COMMENT '订单价格',
  `coupon_id` bigint(50) DEFAULT NULL COMMENT '优惠券ID',
  `coupon_paid` decimal(10,2) DEFAULT NULL COMMENT '优惠券',
  `money_paid` decimal(10,2) DEFAULT NULL COMMENT '已付金额',
  `pay_amount` decimal(10,2) DEFAULT NULL COMMENT '支付金额',
  `add_time` timestamp NULL DEFAULT NULL COMMENT '创建时间',
  `confirm_time` timestamp NULL DEFAULT NULL COMMENT '订单确认时间',
  `pay_time` timestamp NULL DEFAULT NULL COMMENT '支付时间',
  PRIMARY KEY (`order_id`),
  KEY `FK_trade_order` (`user_id`),
  KEY `FK_trade_order2` (`goods_id`),
  KEY `FK_trade_order3` (`coupon_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_order` */

/*Table structure for table `trade_pay` */

DROP TABLE IF EXISTS `trade_pay`;

CREATE TABLE `trade_pay` (
  `pay_id` bigint(50) NOT NULL COMMENT '支付编号',
  `order_id` bigint(50) DEFAULT NULL COMMENT '订单编号',
  `pay_amount` decimal(10,2) DEFAULT NULL COMMENT '支付金额',
  `is_paid` int(1) DEFAULT NULL COMMENT '是否已支付 1否 2是',
  PRIMARY KEY (`pay_id`),
  KEY `FK_trade_pay` (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_pay` */

/*Table structure for table `trade_user` */

DROP TABLE IF EXISTS `trade_user`;

CREATE TABLE `trade_user` (
  `user_id` bigint(50) NOT NULL AUTO_INCREMENT COMMENT '用户ID',
  `user_name` varchar(255) DEFAULT NULL COMMENT '用户姓名',
  `user_password` varchar(255) DEFAULT NULL COMMENT '用户密码',
  `user_mobile` varchar(255) DEFAULT NULL COMMENT '手机号',
  `user_score` int(11) DEFAULT NULL COMMENT '积分',
  `user_reg_time` timestamp NULL DEFAULT NULL COMMENT '注册时间',
  `user_money` decimal(10,0) DEFAULT NULL COMMENT '用户余额',
  PRIMARY KEY (`user_id`)
) ENGINE=InnoDB AUTO_INCREMENT=345963634385633281 DEFAULT CHARSET=utf8;

/*Data for the table `trade_user` */

insert  into `trade_user`(`user_id`,`user_name`,`user_password`,`user_mobile`,`user_score`,`user_reg_time`,`user_money`) values (345963634385633280,'刘备','123L','18888888888L',100,'2019-07-09 13:37:03','900');

/*Table structure for table `trade_user_money_log` */

DROP TABLE IF EXISTS `trade_user_money_log`;

CREATE TABLE `trade_user_money_log` (
  `user_id` bigint(50) NOT NULL COMMENT '用户ID',
  `order_id` bigint(50) NOT NULL COMMENT '订单ID',
  `money_log_type` int(1) NOT NULL COMMENT '日志类型 1订单付款 2 订单退款',
  `use_money` decimal(10,2) DEFAULT NULL,
  `create_time` timestamp NULL DEFAULT NULL COMMENT '日志时间',
  PRIMARY KEY (`user_id`,`order_id`,`money_log_type`),
  KEY `FK_trade_user_money_log2` (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `trade_user_money_log` */

/*!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 */;

3.2 项目初始化

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

3.1.1 工程浏览

在这里插入图片描述

父工程：shop-parent
订单系统：shop-order-web
支付系统：shop-pay-web
优惠券服务：shop-coupon-service
订单服务：shop-order-service
支付服务：shop-pay-service
商品服务：shop-goods-service
用户服务：shop-user-service
实体类：shop-pojo
持久层：shop-dao
接口层：shop-api
工具工程：shop-common

共12个系统

3.1.2 工程关系

在这里插入图片描述

3.3 Mybatis逆向工程使用

1）代码生成

使用Mybatis逆向工程针对数据表生成CURD持久层代码

2）代码导入

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

3.4 公共类介绍

ID生成器

IDWorker：Twitter雪花算法
异常处理类

CustomerException：自定义异常类

CastException：异常抛出类
常量类

ShopCode：系统状态类
响应实体类

Result：封装响应状态和响应信息

4. 下单业务

流程简介*

用户发起下单操作，后台先保存预订单，然后通过RPC调用库存服务，优惠券服务，用户服务完成业务，如果任何1个服务调用失败，那么在处理异常的时候，发送1条消息到RocketMQ，其它服务则监听RocketMQ中的消息，做确认订单失败的业务，回退该预订单的数据。而消息的消费方需要做幂等性处理

下单业务流程图

在这里插入图片描述

4.1 下单基本流程

1）接口定义

IOrderService

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

2）业务类实现

@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）校验订单

在这里插入图片描述

private void checkOrder(TradeOrder order) {
        //1.校验订单是否存在
        if(order==null){
            CastException.cast(ShopCode.SHOP_ORDER_INVALID);
        }
        //2.校验订单中的商品是否存在
        TradeGoods goods = goodsService.findOne(order.getGoodsId());
        if(goods==null){
            CastException.cast(ShopCode.SHOP_GOODS_NO_EXIST);
        }
        //3.校验下单用户是否存在
        TradeUser user = userService.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("校验订单通过");
}

4）生成预订单

在这里插入图片描述

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调用商品服务完成扣减库存

private void reduceGoodsNum(TradeOrder order) {
    
    TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
    
    goodsNumberLog.setGoodsId(order.getGoodsId()); // 商品id
    goodsNumberLog.setOrderId(order.getOrderId()); // 订单id
    goodsNumberLog.setGoodsNumber(order.getGoodsNumber()); // 购买的商品数量
    
    // 通过dubbo的rpc远程调用商品服务实现 扣减库存
    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扣减库存

@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完成扣减优惠券

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更改优惠券状态

@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）扣减用户余额

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

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,更新余额

在这里插入图片描述

@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())    // 用户id
        .andOrderIdEqualTo(userMoneyLog.getOrderId()); // 订单id
    
    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);
    }
    
    //【退款操作】
    else 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）确认订单

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）小结

@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属性值

rocketmq.name-server=192.168.25.135:9876;192.168.25.138:9876
rocketmq.producer.group=orderProducerGroup

mq.order.consumer.group.name=order_orderTopic_cancel_group
mq.order.topic=orderTopic
mq.order.tag.confirm=order_confirm
mq.order.tag.cancel=order_cancel

注入模板类和属性值信息

 @Autowired
 private RocketMQTemplate rocketMQTemplate;

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

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

发送下单失败消息

@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,                                   // topic
                        cancelTag,                               // tag
                        cancelOrderMQ.getOrderId().toString(),   // keys
                        JSON.toJSONString(cancelOrderMQ)         // body
                       );
        } catch (Exception e1) {
            e1.printStackTrace();
            CastException.cast(ShopCode.SHOP_MQ_SEND_MESSAGE_FAIL);
        }
        
        return new Result(ShopCode.SHOP_FAIL.getSuccess(),
                          ShopCode.SHOP_FAIL.getMessage());
    }
}

private void sendMessage(String topic, 
                         String tags, 
                         String keys, 
                         String body) {
    
    //判断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属性值

rocketmq.name-server=192.168.25.135:9876;192.168.25.138:9876
mq.order.consumer.group.name=order_orderTopic_cancel_group
mq.order.topic=orderTopic

创建监听类，消费消息

@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) { // 泛型使用MessageExt, 以获取RocketMQ为消息生成的msgId
        ...
    }
}

1）回退库存*

流程分析
消息消费者

@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);        // tag
            primaryKey.setMsgKey(keys);        // keys
            primaryKey.setGroupName(groupName);// groupName
            
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper
                .selectByPrimaryKey(primaryKey);

            //3. 判断如果消费过...
            if(mqConsumerLog != null){
                
                //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;
                    }
                    
                    // 消费次数未大于3
                    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("并发修改,稍后处理");
                        return;
                    }
                    
                    // 未超过最大允许失败的次数，将接着下面的处理
                }

            } 
            else {
                
                //4. 判断如果没有消费过...
                
                mqConsumerLog = new TradeMqConsumerLog();
                
                // 联合主键
                mqConsumerLog.setGroupName(groupName);
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                
                // 处理中
                mqConsumerLog.setConsumerStatus(
                    ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode()
                );
                
				// 消费次数
                mqConsumerLog.setConsumerTimes(1);

                //将消息处理信息添加到数据库
                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.setGroupName(groupName);
                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{
                // 消费次数 + 1
                mqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes()+1);
                mqConsumerLogMapper.updateByPrimaryKeySelective(mqConsumerLog);
            }
        }

    }
}

2）回退优惠券

@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）回退余额

@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("余额回退失败");
        }

    }
}

updateMoneyPaid如下（但是里面会存在问题，防止多次退款那里没有加锁。假设MQ并发发了2条消息过来，就会给用户加多次记录。可以加上分布式锁，或者使用乐观锁）

@Override
public Result updateMoneyPaid(TradeUserMoneyLog userMoneyLog) {

    //1.校验参数是否合法
    if(userMoneyLog==null || userMoneyLog.getUserId()==null ||
       userMoneyLog.getOrderId()==null || userMoneyLog.getUseMoney()==null||
       userMoneyLog.getUseMoney().compareTo(BigDecimal.ZERO)<=0)
    {
        CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
    }

    //2.查询订单余额使用日志
    TradeUserMoneyLogExample userMoneyLogExample = new TradeUserMoneyLogExample();
    TradeUserMoneyLogExample.Criteria criteria = userMoneyLogExample.createCriteria();
    criteria.andOrderIdEqualTo(userMoneyLog.getOrderId());
    criteria.andUserIdEqualTo(userMoneyLog.getUserId());
    int r = userMoneyLogMapper.countByExample(userMoneyLogExample);

    TradeUser tradeUser = userMapper.selectByPrimaryKey(userMoneyLog.getUserId());

    //3.扣减余额...
    if(userMoneyLog.getMoneyLogType().intValue()
       ==ShopCode.SHOP_USER_MONEY_PAID.getCode().intValue()){

        if(r>0){
            //已经付款
            CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY);
        }

        //减余额
        tradeUser.setUserMoney(new BigDecimal(tradeUser.getUserMoney())
                               .subtract(userMoneyLog.getUseMoney()).longValue()
                              );
        userMapper.updateByPrimaryKey(tradeUser);
    }

    //4.回退余额...
    if(userMoneyLog.getMoneyLogType().intValue()
       ==ShopCode.SHOP_USER_MONEY_REFUND.getCode().intValue()){
        
        if(r<=0){
            //如果没有支付,则不能回退余额
            CastException.cast(ShopCode.SHOP_ORDER_PAY_STATUS_NO_PAY);
        }
        
        //防止多次退款
        TradeUserMoneyLogExample userMoneyLogExample2 = new TradeUserMoneyLogExample();
        TradeUserMoneyLogExample.Criteria criteria1 
                                               = userMoneyLogExample2.createCriteria();
        criteria1.andOrderIdEqualTo(userMoneyLog.getOrderId());
        criteria1.andUserIdEqualTo(userMoneyLog.getUserId());
        criteria1.andMoneyLogTypeEqualTo(ShopCode.SHOP_USER_MONEY_REFUND.getCode());
        
        int r2 = userMoneyLogMapper.countByExample(userMoneyLogExample2);
        
        if(r2>0){
            CastException.cast(ShopCode.SHOP_USER_MONEY_REFUND_ALREADY);
        }
        
        //退款
        tradeUser.setUserMoney(new BigDecimal(tradeUser.getUserMoney())
                               .add(userMoneyLog.getUseMoney()).longValue());
        userMapper.updateByPrimaryKey(tradeUser);
    }
    
    //5.记录订单余额使用日志
    userMoneyLog.setCreateTime(new Date());
    
    userMoneyLogMapper.insert(userMoneyLog);
    
    return new Result(ShopCode.SHOP_SUCCESS.getSuccess(),
                      ShopCode.SHOP_SUCCESS.getMessage());
}

4）取消订单

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

    @Autowired
    private TradeOrderMapper orderMapper;

    @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("接受消息成功");
            
            //2. 查询订单
            TradeOrder order = orderMapper.selectByPrimaryKey(mqEntity.getOrderId());
            
            //3.更新订单状态为取消
            order.setOrderStatus(ShopCode.SHOP_ORDER_CANCEL.getCode());
            
            orderMapper.updateByPrimaryKey(order);
            
            log.info("订单状态设置为取消");
            
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            log.info("订单取消失败");
        }
    }
}

4.3 测试

1）准备测试环境

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

    @Autowired
    private IOrderService orderService;
}

2）准备测试数据

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

3）测试下单成功流程

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

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

4）测试下单失败流程

代码同上。

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

5. 支付业务

流程简介*

用户对订单发起支付，创建支付订单，用户重定向到第三方支付平台，当用户在第三方平台支付成功后，第三方支付平台发起支付成功的回调，后台需要及时给第三方支付平台响应，因此在收到回调后，先保存支付成功的消息到数据库以保证消息的可靠性，然后通过线程池将消息异步发送到RocketMQ，其它服务监听RocketMQ做支付成功的逻辑。

5.1 创建支付订单

在这里插入图片描述

// 已知 订单id, 需支付金额, 未支付状态, 来保存 支付订单
public Result createPayment(TradePay tradePay) {
    
    // 必须携带orderId
    if(tradePay==null || tradePay.getOrderId()==null){
        CastException.cast(ShopCode.SHOP_REQUEST_PARAMETER_VALID);
    }
    
    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();
        // 设置支付id
        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 代码实现

// 用户通过第三方支付平台, 支付成功后, 由支付平台回调此接口

// TradePay中定义的属性: payId、orderId、payAmount、isPaid、
public Result callbackPayment(TradePay tradePay) {

    // 1. 判断用户支付状态
    // 如果已经支付了
    if (tradePay.getIsPaid().equals(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode())) {

        // 2. 更新支付订单状态为已支付

        // 判断支付订单是否存在, 如果不存在, 则抛出异常
        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) {

            // 3. 创建支付成功的消息
            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());

            // 4. 将支付成功的消息持久化数据库
            mqProducerTempMapper.insert(mqProducerTemp);

            TradePay finalTradePay = tradePay;

            // 在线程池中进行处理
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    
                    try {
                        
                        // 5. 发送消息到rocketmq
                        SendResult sendResult = sendMessage( 
                                                       topic, // payTopic
                                                       tag,   // paid
                                                       finalTradePay.getPayId(),
                                                       JSON.toJSONString( finalTradePay)
                                                );
                        
                        log.info(JSON.toJSONString(sendResult));
                        
                        if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) { 
                            
                            // 6. 等待发送结果,如果MQ接受到消息,删除发送成功的消息
                            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());
}

/**
  * 发送支付成功消息
  * @param topic
  * @param tag
  * @param key
  * @param body
  */
private SendResult sendMessage(String topic, 
                               String tag, 
                               String key, 
                               String body) throws Exception{
    
    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,tag,key,body.getBytes());
    
    SendResult sendResult = rocketMQTemplate.getProducer().send(message);
    
    return sendResult;
}

线程池优化消息发送逻辑

创建线程池对象

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

}

使用线程池

@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属性值

mq.pay.topic=payTopic
mq.pay.consumer.group.name=pay_payTopic_group

2）消费消息

在订单服务中，配置公共的消息处理类

public class BaseConsumer {

    public TradeOrder handleMessage(IOrderService 
                                    orderService, 
                                    MessageExt messageExt,Integer code) throws Exception {
        //解析消息内容
        String body = new String(messageExt.getBody(), "UTF-8");
        String msgId = messageExt.getMsgId();
        String tags = messageExt.getTags();
        String keys = messageExt.getKeys();
        OrderMQ orderMq = JSON.parseObject(body, OrderMQ.class);
        
        //查询
        TradeOrder order = orderService.findOne(orderMq.getOrderId());

        if(ShopCode.SHOP_ORDER_MESSAGE_STATUS_CANCEL.getCode().equals(code)){
            order.setOrderStatus(ShopCode.SHOP_ORDER_CANCEL.getCode());
        }

        if(ShopCode.SHOP_ORDER_MESSAGE_STATUS_ISPAID.getCode().equals(code)){
            order.setPayStatus(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
        }
        orderService.changeOrderStatus(order);
        return order;
    }

}

接受订单支付成功消息

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

    @Autowired
    private TradeOrderMapper orderMapper;

    @Override
    public void onMessage(MessageExt messageExt) {

        log.info("接收到支付成功消息");

        try {
            
            //1.解析消息内容
            String body = new String(messageExt.getBody(),"UTF-8");
            TradePay tradePay = JSON.parseObject(body,TradePay.class);
            
            //2.根据订单ID查询订单对象
            TradeOrder tradeOrder =                  
                          orderMapper.selectByPrimaryKey(tradePay.getOrderId());
            
            //3.更改订单支付状态为已支付
            tradeOrder.setPayStatus(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
            
            //4.更新订单数据到数据库
            orderMapper.updateByPrimaryKey(tradeOrder);
            
            log.info("更改订单支付状态为已支付");
            
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

    }
}

6. 整体联调

通过Rest客户端请求shop-order-web和shop-pay-web完成下单和支付操作

6.1 准备工作

1）配置RestTemplate类

@Configuration
public class RestTemplateConfig {

    @Bean
    @ConditionalOnMissingBean({ RestOperations.class, RestTemplate.class })
    public RestTemplate restTemplate(ClientHttpRequestFactory factory) {

        RestTemplate restTemplate = new RestTemplate(factory);
        

        // 使用 utf-8 编码集的 conver 替换默认的 conver
        // （默认的 string conver 的编码集为"ISO-8859-1"）
        List<HttpMessageConverter<?>> 
            messageConverters = restTemplate.getMessageConverters();
        
        Iterator<HttpMessageConverter<?>> iterator = messageConverters.iterator();
        
        while (iterator.hasNext()) {
            HttpMessageConverter<?> converter = iterator.next();
            if (converter instanceof StringHttpMessageConverter) {
                iterator.remove();
            }
        }
        
        messageConverters.add(new StringHttpMessageConverter(Charset.forName("UTF-8")));

        return restTemplate;
    }

    @Bean
    @ConditionalOnMissingBean({ClientHttpRequestFactory.class})
    public ClientHttpRequestFactory simpleClientHttpRequestFactory() {
        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
        // ms
        factory.setReadTimeout(15000);
        // ms
        factory.setConnectTimeout(15000);
        return factory;
    }
}

2）配置请求地址

订单系统

server.host=http://localhost
server.servlet.path=/order-web
server.port=8080
shop.order.baseURI=${server.host}:${server.port}${server.servlet.path}
shop.order.confirm=/order/confirm

支付系统

server.host=http://localhost
server.servlet.path=/pay-web
server.port=9090
shop.pay.baseURI=${server.host}:${server.port}${server.servlet.path}
shop.pay.createPayment=/pay/createPayment
shop.pay.callbackPayment=/pay/callbackPayment

6.2 下单测试

@RunWith(SpringRunner.class)
@ContextConfiguration(classes = ShopOrderWebApplication.class)
@TestPropertySource("classpath:application.properties")
public class OrderTest {

   @Autowired
   private RestTemplate restTemplate;

   @Value("${shop.order.baseURI}")
   private String baseURI;

   @Value("${shop.order.confirm}")
   private String confirmOrderPath;

   @Autowired
   private IDWorker idWorker;
  
  /**
    * 下单
    */
   @Test
   public void confirmOrder(){
       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));

       Result result = restTemplate.postForEntity(baseURI + confirmOrderPath, 
                                                  order, Result.class).getBody();
       System.out.println(result);
   }

}

6.3 支付测试

@RunWith(SpringRunner.class)
@ContextConfiguration(classes = ShopPayWebApplication.class)
@TestPropertySource("classpath:application.properties")
public class PayTest {

    @Autowired
    private RestTemplate restTemplate;

    @Value("${shop.pay.baseURI}")
    private String baseURI;

    @Value("${shop.pay.createPayment}")
    private String createPaymentPath;

    @Value("${shop.pay.callbackPayment}")
    private String callbackPaymentPath;

    @Autowired
    private IDWorker idWorker;

   /**
     * 已知订单id和需支付的金额, 来创建支付订单
     */
    @Test
    public void createPayment(){

        Long orderId = 346321587315814400L;
        
        TradePay pay = new TradePay();
        pay.setOrderId(orderId);
        pay.setPayAmount(new BigDecimal(4800));

        Result result = restTemplate.postForEntity(baseURI + createPaymentPath,
                                                   pay, Result.class
                                                  )
                                    .getBody();
        
        System.out.println(result);
    }
   
    /**
     * 用户通过第三方支付平台支付成功后, 第三方支付平台回调此接口
     * 支付回调
     */
    @Test
    public void callbackPayment(){
        
        Long payId = 346321891507720192L;
        
        TradePay pay = new TradePay();
        pay.setPayId(payId);
        pay.setIsPaid(ShopCode.SHOP_ORDER_PAY_STATUS_IS_PAY.getCode());
        
        Result result = restTemplate.postForEntity(baseURI + callbackPaymentPath, 
                                                   pay, Result.class).getBody();
        
        System.out.println(result);

    }

}