RabbitMQ实现分布式事务,保证数据一致性
共 14228字,需浏览 29分钟
·
2021-09-12 21:28
点击“蓝字”,关注,置顶公众号
每日技术干货,第一时间送达!
一、实验环境
Lunix系统:Centos7.5 安装软件:rabbitmq 开发工具:IDEA
二、实验目的
Rabbitmq实现多系统间的分布式事务,保证数据一致性
三、实验方案
rabbitmq作为消息中间件
订单中心和运单中心分别作为消息的生产者和消息的消费者,通过rabbitmq传递消息
订单中心作为生产者,模拟用户创建订单,在本地持久化订单信息,记录消息的状态信息,并将消息发送到rabbitmq,同时开启confirm机制,接收消息中间件rabbitmq的响应信息,更新本地消息发送状态(定时任务轮训消息状态信息表,一定时间内未发送成功的数据将再次发起推送,保证atlestonce.
运单中心作为消费者,消费rabbitmq中的订单信息,开启ack确认机制,确保不遗漏订单。并通过消息全局唯一ID保证数据的唯一性,不重复处理订单。
四、实验步骤
1、消息队列
1.1 rabbitmq安装过程略过。。。。
1.2 创建订单交换器:orderExchange
![](https://filescdn.proginn.com/81b06e3a6b4bf419ad5bd6322b3a55fd/27b613ed80ea0d2b87b0e9d58267dfd7.webp)
1.3 创建订单队列:orderQueue
![](https://filescdn.proginn.com/b3a77bcf71f0925d566969e24675ddaf/3399a6628668302cc6e6edc6e8913c6e.webp)
1.4 绑定
![](https://filescdn.proginn.com/881a9c2130ec490556a8a81d7387812d/31b8fc0ea4d6e72f9ace294471097291.webp)
2、数据库准备
2.1订单表
![](https://filescdn.proginn.com/6db76ba5402d4710e70249afbc5cd65a/330af2ca125cee1862515f4e23886fe7.webp)
2.2 消息发送状态表
![](https://filescdn.proginn.com/7a64dc98a70501ffe8849db5e53e0f6b/41062020792d02f681f6547f1c99e0a2.webp)
2.3 运单表
![](https://filescdn.proginn.com/6fdf547a9ac3245b357976733365621f/438961cfcbde2b582f7e8d730b328cf0.webp)
3、订单中心
3.1 订单中心分析
![](https://filescdn.proginn.com/d2110fc0f2f66ba51734ae9068cab870/fe4c9e93db41fac372af4e01f983addc.webp)
利用Rabbitmq发布确认机制(confirm),确保发送成功的数据能被通知到 做个定时任务轮训发送失败以及发送后未响应的订单信息,重新发送。推荐:Java进阶视频资源
3.2 编写代码
3.2.1 Springboot整合rabbitmq和mysql数据库
3.2.1.1依赖如下:
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-webartifactId>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-amqpartifactId>
dependency>
<dependency>
<groupId>mysqlgroupId>
<artifactId>mysql-connector-javaartifactId>
<scope>runtimescope>
dependency>
<dependency>
<groupId>org.projectlombokgroupId>
<artifactId>lombokartifactId>
<optional>trueoptional>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-jdbcartifactId>
dependency>
<dependency>
<groupId>com.alibabagroupId>
<artifactId>fastjsonartifactId>
<version>1.2.17version>
dependency>
3.2.1.2配置文件内容:
server:
port: 8080
spring:
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8&allowMultiQueries=true
username: root
password: root123
rabbitmq:
host: localhost
port: 5672
username: admin
password: admin123
virtual-host: /
#必须配置这个,生产者才会确认回调
publisher-confirm-type: correlated
publisher-returns: true
#重要,手动开启消费者ACK,控制消息在MQ中的删除、重发
listener:
simple:
acknowledge-mode: MANUAL
3.2.2 订单中心代码
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.beans.Transient;
/**
* @Author Lee
* @Description 订单中心
* @Date 2020/1/30 16:57
* @Version 1.0
*/
@Slf4j
@Service
public class OrderService {
@Autowired
private JdbcTemplate jdbcTemplate;
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void setup(){
//消息发送完成后,则回调此方法,ack代表此方法是否发送成功
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
//ack为true,代表MQ已经准确收到消息
if(!ack){
return;
}
try{
String sql = "update tb_msgstatus set status = 1 where msgid = ?";
int count = jdbcTemplate.update(sql,correlationData.getId());
if(count != 1){
log.warn("本地消息表状态修改失败");
}
}catch (Exception e){
log.warn("本息消息表状态修改异常",e);
}
}
});
}
/**
* 创建订单信息
* @param order 订单信息
* @throws Exception
*/
public void createOrder(JSONObject order) throws Exception {
//保存订单信息
saveOrder(order);
//发送MQ消息,直接发送时不可靠,可能会失败(发送后根据回执修改状态表,定时任务扫表读取失败数据重新发送)
sendMsg(order);
}
/**
* 发送订单信息至MQ
* @param order 订单信息
*/
private void sendMsg(JSONObject order) {
//发送消息到MQ,CorrelationData作用:当收到消息回执时会带上这个参数
rabbitTemplate.convertAndSend("orderExchange","",order.toJSONString(),new CorrelationData((String) order.get("orderid")));
}
/**
* 保存订单信息
* @param order 订单信息
* @throws Exception
*/
@Transient
private void saveOrder(JSONObject order) throws Exception {
String sql = "insert into tb_order (orderid,userid,goodsid,ordertime) values (? , ? , ? , now())";
//保存订单信息
int count = jdbcTemplate.update(sql,order.get("orderid"),order.get("userid"),order.get("goodsid"));
if(count != 1){
throw new Exception("订单创建失败");
}
//保存消息发送状态
saveLocalMsg(order);
}
/**
* 记录消息发送状态
* @param order 订单信息
* @throws Exception
*/
private void saveLocalMsg(JSONObject order) throws Exception {
String sql = "insert into tb_msgstatus (msgid,msg,status,sendtime) values (? , ? , 0 , now())";
//记录消息发送状态
int count = jdbcTemplate.update(sql,order.get("orderid"),order.toJSONString());
if(count != 1){
throw new Exception("记录消息发送状态失败");
}
}
}
3.3 订单中心测试
3.3.1 测试代码
@Autowired
private OrderService orderService;
@Test
public void orderServiceTest() throws Exception {
//生成订单信息
JSONObject orderinfo = new JSONObject();
orderinfo.put("orderid",UUID.randomUUID().toString());
orderinfo.put("userid",UUID.randomUUID().toString());
orderinfo.put("goodsid",UUID.randomUUID().toString());
orderService.createOrder(orderinfo);
}
3.3.2 测试验证结果
orderQueue消息队列中已经接收到数据
![](https://filescdn.proginn.com/8c800d2260bae78ea7619f8dec786f6b/52e978beff9a744e77d1c25c2cef98f7.webp)
![](https://filescdn.proginn.com/ad6462ca24172f48cf3fc49f66ce53e4/b28fcbbc68a92cb025e3d1c376634faf.webp)
订单表里的数据
![](https://filescdn.proginn.com/249a5555e4fb8c47d4b7293f3411b101/f2bd82fd238a653a63b313c8692ce889.webp)
状态表数据:
![](https://filescdn.proginn.com/92e4b8150fdce5d20ac26a31cc64469e/aae287543c1de8cab393ec2a5e31461c.webp)
4、运单中心
4.1 运单中心分析
![](https://filescdn.proginn.com/4dd33de185906bec5b126ffa16ea7fb5/63c3cac6828c8273e778515c998a59fb.webp)
消费者收到消息进行处理,处理成功则发送ACK消息通知MQ清除该条记录,否则通知MQ重发或者等待MQ自动重发。本地维护一个处理次数,如果多次处理仍然失败,则将该消息丢弃或者加入到死信队列(DLQ)中。死信队列中的数据可以人工干预。推荐:Java进阶视频资源
4.2 编写代码
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import java.beans.Transient;
import java.io.IOException;
/**
* @Author Lee
* @Description 运单系统
* @Date 2020/1/30 21:58
* @Version 1.0
*/
@Slf4j
@Service
public class DispatchService {
@Autowired
private JdbcTemplate jdbcTemplate;
@RabbitListener(queues = "orderQueue")
public void messageCunsumer(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try{
//MQ里面的数据转换成JSON数据
JSONObject orderInfo = JSONObject.parseObject(message);
log.warn("收到MQ里面的消息:" + orderInfo.toJSONString());
Thread.sleep(1000L);
//执行业务操作,同一个数据不能处理两次,根据业务情况去重,保证幂等性
String orderid = orderInfo.getString("orderid");
//分配快递员配送
dispatch(orderid);
//ack 通知MQ数据已经收到
channel.basicAck(tag,false);
}catch (Exception e){
//异常情况,需要根据需求去重发或者丢弃
//重发一定次数后丢弃,日志告警(rabbitmq没有设置重发次数功能,重发时需要代码实现,比如使用redis记录重发次数,)
channel.basicNack(tag,false,false);
//系统关键数据异常,需要人工干预
}
//如果不给确认回复,就等这个consumer断开连接后,MQ会继续推送
}
/**
* 分配快递员
* @param orderid 订单编号
*/
@Transient
private void dispatch(String orderid) throws Exception {
String sql = "insert into tb_dispatch (orderid,courier,status) values (?,?,?)";
int count = jdbcTemplate.update(sql,orderid,"东哥","配送中");
if(count != 1){
throw new Exception("调度数据插入失败,原因[数据库操作]");
}
}
}
4.3 订单中心测试
启动springboot后自动监听MQ中的消息队列,自动处理
测试结果如下:
![](https://filescdn.proginn.com/d9e8b73f141c9795ded3c11077cbc39c/9de8719ec22f1bf32185517725761d75.webp)
感谢阅读,希望对你有所帮助 :)
来源:blog.csdn.net/qq_31463999/article/details/79220061