目录
可靠消息最终一致性分布式事务实战_Docker安装 RocketMQ
可靠消息最终一致性分布式事务实现_RocketMQ事务消息
RocketMQ是阿里巴巴开源的一款支持事务消息的消息中间件，于 2012年正式开源，2017年成为Apache基金会的顶级项目。
实现原理
RocketMQ 4.3版之后引入了完整的事务消息机制，其内部实现了完 整的本地消息表逻辑，使用RocketMQ实现可靠消息分布式事务就 不用用户再实现本地消息表的逻辑了，极大地减轻了开发工作量。
可靠消息最终一致性分布式事务实战_案列业务介绍
业务介绍
通过RocketMQ中间件实现可靠消息最终一致性分布式事务，模拟 商城业务中的下单扣减库存场景。订单微服务和库存微服务分别独立开发和部署。
流程
架构选型
数据库表设计
orders订单数据表
orders数据表存储于tx-msg-orders订单数据库。
DROP TABLE IF EXISTS `orders`;
CREATE TABLE `order` (
`id` bigint(20) NOT NULL COMMENT '主键',
`create_time` datetime(0) NULL DEFAULT NULL COMMENT '创建时间',
`order_no` varchar(64) CHARACTER SET utf8
COLLATE utf8_bin NULL DEFAULT NULL COMMENT '订单
编号',
`product_id` bigint(20) NULL DEFAULT NULL COMMENT '商品id',
`pay_count` int(11) NULL DEFAULT NULL COMMENT '购买数量',
PRIMARY KEY (`id`) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE
= utf8_bin ROW_FORMAT = Dynamic;
SET FOREIGN_KEY_CHECKS = 1;
CREATE TABLE `tx_log` (
`tx_no` varchar(64) CHARACTER SET utf8
COLLATE utf8_bin NOT NULL COMMENT '分布式事务全局序列号',
`create_time` datetime(0) NULL DEFAULT NULL COMMENT '创建时间',
PRIMARY KEY (`tx_no`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;stock库存数据表
DROP TABLE IF EXISTS `stock`;
CREATE TABLE `stock` (
`id` bigint(20) NOT NULL COMMENT '主键id',
`product_id` bigint(20) NULL DEFAULT NULL COMMENT '商品id',
`total_count` int(11) NULL DEFAULT NULL COMMENT '商品总库存',
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
-- ----------------------------
-- Table structure for tx_log
-- ----------------------------
DROP TABLE IF EXISTS `tx_log`;
CREATE TABLE `tx_log` (
`tx_no` varchar(64) CHARACTER SET utf8
COLLATE utf8_bin NOT NULL COMMENT '分布式事务全局序列号',
`create_time` datetime(0) NULL DEFAULT NULL COMMENT '创建时间',
PRIMARY KEY (`tx_no`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
SET FOREIGN_KEY_CHECKS = 1;tx_log事务记录表
可靠消息最终一致性分布式事务实战_Docker安装 RocketMQ
在安装RocketMQ之前，我们先了解一下RocketMQ的部署架构，了 解一下RocketMQ的组件，然后基于当前主流的Docker安装 RocketMQ，我们这里安装单台RocketMQ，但为了防止单节点故 障、保障高可用，生产环境建议安装RocketMQ集群。
安装NameServer
拉取镜像
docker pull rocketmqinc/rocketmq创建数据存储目录
mkdir -p /docker/rocketmq/data/namesrv/logs
/docker/rocketmq/data/namesrv/store启动NameServer
docker run -d \
--restart=always \
--name rmqnamesrv \
-p 9876:9876 \
-v /docker/rocketmq/data/namesrv/logs:/root/logs \
-v /docker/rocketmq/data/namesrv/store:/root/store \
-e "MAX_POSSIBLE_HEAP=100000000" \
rocketmqinc/rocketmq \
sh mqnamesrv
安装Broker
border配置：创建 broker.conf 配置文件
vim /docker/rocketmq/conf/broker.conf
# 所属集群名称，如果节点较多可以配置多个
brokerClusterName = DefaultCluster
#broker名称，master和slave使用相同的名称，表明他们的
主从关系
brokerName = broker-a
#0表示Master，大于0表示不同的
slave brokerId = 0
#表示几点做消息删除动作，默认是凌晨4点
deleteWhen = 04
#在磁盘上保留消息的时长，单位是小时
fileReservedTime = 48
#有三个值：SYNC_MASTER，ASYNC_MASTER，SLAVE；同步和
异步表示Master和Slave之间同步数据的机 制；
brokerRole = ASYNC_MASTER
#刷盘策略，取值为：ASYNC_FLUSH，SYNC_FLUSH表示同步刷
盘和异步刷盘；SYNC_FLUSH消息写入磁盘后 才返回成功状
态，ASYNC_FLUSH不需要；
flushDiskType = ASYNC_FLUSH
# 设置broker节点所在服务器的ip地址
brokerIP1 = 192.168.66.100
#剩余磁盘比例
diskMaxUsedSpaceRatio=99启动broker
docker run -d --restart=always --name rmqbroker
--link rmqnamesrv:namesrv -p 10911:10911 -p
10909:10909 --privileged=true -v
/docker/rocketmq/data/broker/logs:/root/logs -v
/docker/rocketmq/data/broker/store:/root/store
-v
/docker/rocketmq/conf/broker.conf:/opt/rocketmq
-4.4.0/conf/broker.conf -e
"NAMESRV_ADDR=namesrv:9876" -e
"MAX_POSSIBLE_HEAP=200000000"
rocketmqinc/rocketmq sh mqbroker -c
/opt/rocketmq-4.4.0/conf/broker.conf
报错：
部署RocketMQ的管理工具
RocketMQ提供了UI管理工具，名为rocketmq-console，我们选择 docker安装
#创建并启动容器
docker run -d --restart=always --name rmqadmin
-e "JAVA_OPTS=-
Drocketmq.namesrv.addr=192.168.66.100:9876 -
Dcom.rocketmq.sendMessageWithVIPChannel=false"
-p 8080:8080 pangliang/rocketmq-console-ng关闭防火墙(或者开放端口)
#关闭防火墙
systemctl stop firewalld.service
#禁止开机启动
systemctl disable firewalld.service测试
访问：http://192.168.66.101:8080/#/ （可以切换中文）
可靠消息最终一致性分布式事务实战_实现订单微服务
创建父工程rocketmq-msg
创建订单微服务子工程
引入依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starterweb</artifactId>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connectorjava</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-bootstarter</artifactId>
<version>2.0.2</version>
</dependency>
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-bootstarter</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>编写配置文件
server:
port: 9090
spring:
application:
name: tx-msg-stock
datasource:
url: jdbc:mysql://192.168.66.100:3306/txmsg-order?
useUnicode=true&characterEncoding=UTF8&useOldAliasMetadataBehavior=true&autoReconnec
t=true&failOverReadOnly=false&useSSL=false
username: root
password: 123456
driver-class-name: com.mysql.cj.jdbc.Driver
################ RocketMQ 配置 ##########
rocketmq:
name-server: 192.168.66.100:9876
producer:
group: order-group编写主启动类
/**
* 订单微服务启动成功
*/
@Slf4j
@MapperScan("com.tong.order.mapper")
@SpringBootApplication
public class OrderMain9090 {
public static void main(String[] args) {
SpringApplication.run(OrderMain9090.class,args);
log.info("************* 订单微服务启动成功*******");
}
}代码生成
package com.tong.utils;
import com.baomidou.mybatisplus.generator.FastAutoGenerator;
import com.baomidou.mybatisplus.generator.config.rules.NamingStrategy;
import java.util.Arrays;
import java.util.List;
public class CodeGenerator {
public static void main(String[] args) {
FastAutoGenerator.create("jdbc:mysql://192.168.66.102:3306/tx-msg-order", "root", "123456")
.globalConfig(builder -> {
builder.author("tong")// 设置作者
.commentDate("MMdd") // 注释日期格式
.outputDir(System.getProperty("user.dir")+"/rocketmq-msg/orders"+ "/src/main/java/")
.fileOverride(); //覆盖文件
})
// 包配置
.packageConfig(builder -> {
builder.parent("com.tong.orders") // 包名前缀
.entity("entity")//实体类包名
.mapper("mapper")//mapper接口包名
.service("service"); //service包名
})
.strategyConfig(builder -> {
// 设置需要生成的表名
builder.addInclude(Arrays.asList("orders","tx_log"))
// 开始实体类配置
.entityBuilder()
// 开启lombok模型
.enableLombok()
//表名下划线转驼峰
.naming(NamingStrategy.underline_to_camel)
//列名下划线转驼峰
.columnNaming(NamingStrategy.underline_to_camel);
})
.execute();
}
}创建TxMessage类
在项目的com.itbaizhan.orders.tx包下创建TxMessage类，主要用 来封装实现分布式事务时，在订单微服务、RocketMQ消息中间件 和库存微服务之间传递的全局事务消息，项目中会通过事务消息实现幂等。
@Data
@NoArgsConstructor
@AllArgsConstructor
public class TxMessage implements Serializable
{
private static final long serialVersionUID = -4704980150056885074L;
/**
* 商品id
*/
private Long productId;
/**
* 商品购买数量
*/
private Integer payCount;
/**
* 全局事务编号
*/
private String txNo;
}可靠消息最终一致性分布式事务实战_订单微服务业务层实现
业务逻辑层主要实现了用户提交订单后的业务逻辑。
编写OrderService接口
/**
* 添加订单
* @param productId 商品id
* @param payCount 购买数量
*/
void save(Long productId,Integer payCount);
/**
* 提交订单同时保存事务信息
*/
void submitOrderAndSaveTxNo(TxMessage txMessage);
/**
* 提交订单
* @param productId 商品id
* @param payCount 购买数量
*/
void submitOrder(Long productId, Integer payCount);编写OrderService接口实现
package com.itbaizhan.order.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.tong.order.entity.Order;
import com.tong.order.entity.TxLog;
import com.tong.order.mapper.OrderMapper;
import com.tong.order.mapper.TxLogMapper;
import com.tong.order.service.IOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tong.order.tx.TxMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.UUID;
/**
* <p>
* 服务实现类
* </p>
*
* @author tong
* @since 05-20
*/
@Slf4j
@Service
public class OrderServiceImpl extends
ServiceImpl<OrderMapper, Order> implements IOrderService {
@Resource
RocketMQTemplate rocketMQTemplate;
@Resource
private TxLogMapper txLogMapper;
/**
* 添加
* @param productId 商品id
* @param payCount 购买数量
*/
@Override
public void save(Long productId, Integer payCount) {
Order order = new Order();
// 订单创建时间
order.setCreateTime(LocalDateTime.now());
// 生产订单编号
order.setOrderNo(UUID.randomUUID().toString().replace("-",""));
// 商品id
order.setProductId(productId);
// 购买数量
order.setPayCount(payCount);
baseMapper.insert(order);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void submitOrderAndSaveTxNo(TxMessage txMessage) {
TxLog txLog = txLogMapper.selectById(txMessage.getTxNo());
if(txLog != null){
log.info("订单微服务已经执行过事务,商品id为:{}，事务编号为:{}",txMessage.getProductId(),txMessage.getTxNo());
return;
}
//生成订单
this.save(txMessage.getProductId(),txMessage.getPayCount());
//生成订单
txLog = new TxLog();
txLog.setTxNo(txMessage.getTxNo());
txLog.setCreateTime(LocalDateTime.now());
//添加事务日志
txLogMapper.insert(txLog);
}
/**
* 提交订单
* @param productId 商品id
* @param payCount 购买数量
*/
@Override
public void submitOrder(Long productId,Integer payCount) {
//生成全局分布式序列号
String txNo = UUID.randomUUID().toString();
TxMessage txMessage = new TxMessage(productId, payCount, txNo);
JSONObject jsonObject = new JSONObject();
jsonObject.put("txMessage", txMessage);
Message<String> message = MessageBuilder.withPayload(jsonObject.toJSONString()).build();
//发送事务消息 且该消息不允许消费
tx_order_group: 指定版事务消息组
rocketMQTemplate.sendMessageInTransaction("tx_order_group", "topic_txmsg", message, null);
}
}
可靠消息最终一致性分布式事务实战_订单微服务监听事务消息
执行本地的业务代码
package com.tong.order.message;
import com.alibaba.fastjson.JSONObject;
import com.tong.order.entity.TxLog;
import com.tong.order.mapper.TxLogMapper;
import com.tong.order.service.IOrderService;
import com.tong.order.service.ITxLogService;
import com.tong.order.tx.TxMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
/**
* @author tong
* @version 1.0.0
* @description 监听事务消息
*/
@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "tx_order_group")
public class OrderTxMessageListener implements
RocketMQLocalTransactionListener {
@Autowired
private IOrderService orderService;
@Resource
private TxLogMapper txLogMapper;
/**
* RocketMQ的Producer本地事务：先执行本地的业务代码（使用Spring的事件管理），判断是否成功。
* 成功返回： RocketMQLocalTransactionState.COMMIT,
* 失败返回：RocketMQLocalTransactionState.ROLLBACK
*/
@Override
@Transactional(rollbackFor = Exception.class)
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object obj)
{
try {
log.info("订单微服务执行本地事务");
TxMessage txMessage = this.getTxMessage(msg);
//执行本地事务
orderService.submitOrderAndSaveTxNo(txMessage);
//提交事务
log.info("订单微服务提交事务");
// COMMIT:即生产者通知Rocket该消息可以消费
return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
e.printStackTrace();
//异常回滚事务
log.info("订单微服务回滚事务");
// ROLLBACK:即生产者通知Rocket将该消息删除
return RocketMQLocalTransactionState.ROLLBACK;
}
}
private TxMessage getTxMessage(Message msg)
{
String messageString = new String((byte[]) msg.getPayload());
JSONObject jsonObject = JSONObject.parseObject(messageString);
String txStr = jsonObject.getString("txMessage");
return JSONObject.parseObject(txStr,TxMessage.class);
}
}网络异常消息处理
/**
* 因为网络异常或其他原因时，RocketMQ的消息状态处于UNKNOWN时，调用该方法检查Producer的本地
* 事务是否已经执行成功,
* 成功返回： RocketMQLocalTransactionState.COMMIT,
* 失败返回：RocketMQLocalTransactionState.ROLLBACK
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
log.info("订单微服务查询本地事务");
TxMessage txMessage = this.getTxMessage(msg);
// 获取订单的消息
Integer exists = txLogService.isExistsTx(txMessage.getTxNo());
if (exists != null) {
// COMMIT:即生产者通知Rocket该消息可以消费
return RocketMQLocalTransactionState.COMMIT;
}
// UNKNOWN:即生产者通知Rocket继续查询该消息的状态
return RocketMQLocalTransactionState.UNKNOWN;
}
private TxMessage getTxMessage(Message msg)
{
String messageString = new String((byte[]) msg.getPayload());
JSONObject jsonObject = JSONObject.parseObject(messageString);
String txStr = jsonObject.getString("txMessage");
return JSONObject.parseObject(txStr,TxMessage.class);
}可靠消息最终一致性分布式事务实战_实现库存微服务
创建库存微服务tx-msg-stock
引入依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starterweb</artifactId>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connectorjava</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-bootstarter</artifactId>
<version>2.0.1</version>
</dependency>
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-bootstarter</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>编写配置文件
server:
port: 6060
spring:
application:
name: tx-msg-stock
datasource:
url: jdbc:mysql://192.168.66.100:3306/txmsg-stock?
useUnicode=true&characterEncoding=UTF8&useOldAliasMetadataBehavior=true&autoReconnec
t=true&failOverReadOnly=false&useSSL=false
username: root
password01: 123456
driver-class-name: com.mysql.cj.jdbc.Driver
################ RocketMQ 配置 ##########
rocketmq:
name-server: 192.168.66.100:9876编写主启动类
package com.tong.stock;
import lombok.extern.slf4j.Slf4j;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author tong
* @version 1.0.0
* @description 库存微服务启动类
*/
@MapperScan("com.tong.stock.mapper")
@Slf4j
@SpringBootApplication
public class StockServerStarter {
public static void main(String[] args) {
SpringApplication.run(StockServerStarter.class, args);
log.info("**************** 库存服务启动成功 ***********");
}
}代码生成
package com.tong.utils;
import com.baomidou.mybatisplus.generator.FastAutoGenerator;
import com.baomidou.mybatisplus.generator.config.rules.NamingStrategy;
import java.util.Arrays;
import java.util.List;
public class CodeGenerator {
public static void main(String[] args) {
FastAutoGenerator.create("jdbc:mysql://192.168.66.102:3306/tx-msg-stock", "root", "123456")
.globalConfig(builder -> { builder.author("tong")// 设置作者
.commentDate("MMdd") // 注释日期格式
.outputDir(System.getProperty("user.dir") +"/rocketmq-msg/stock"+ "/src/main/java/")
.fileOverride(); //覆盖文件
})
// 包配置
.packageConfig(builder -> {
builder.parent("com.tong.stock") // 包名前缀
.entity("entity")//实体类包名
.mapper("mapper")//mapper接口包名
.service("service"); //service包名
})
.strategyConfig(builder -> {
// 设置需要生成的表名
builder.addInclude(Arrays.asList("stock","tx_log"))
// 开始实体类配置
.entityBuilder()
// 开启lombok模型
.enableLombok() //表名下划线转驼峰
.naming(NamingStrategy.underline_to_camel)
//列名下划线转驼峰
.columnNaming(NamingStrategy.underline_to_camel);
})
.execute();
}
}编写库存接口
public interface StockService {
/**
* 根据id查询库存
* @param id
* @return
*/
Stock getStockById(Long id);
/**
* 扣减库存
*/
void decreaseStock(TxMessage txMessage);
}