Rabbitmq实现多系统间的分布式事务，保证数据一致性
- 一、实验环境
- 二、实验目的
- 三、实验方案
- 四、实验步骤
- 1、消息队列
- 1.1 rabbitmq安装过程略过。。。。
- 1.2 创建订单交换器：orderExchange
- 1.3 创建订单队列：orderQueue
- 1.4 绑定
- 2、数据库准备
- 2.1订单表
- 2.2 消息发送状态表
- 2.3 运单表
- 3、订单中心
- 3.1 订单中心分析
- 3.2 编写代码
- 3.2.1 Springboot整合rabbitmq和mysql数据库
- 3.2.1.1依赖如下：
- 3.2.1.2配置文件内容：
- 3.2.2 订单中心代码
- 3.3 订单中心测试
- 3.3.1 测试代码
- 3.3.2 测试验证结果
- 4、运单中心
- 4.1 运单中心分析
- 4.2 编写代码
- 4.3 订单中心测试
一、实验环境
Lunix系统：Centos7.5
安装软件：rabbitmq
开发工具：IDEA
二、实验目的
Rabbitmq实现多系统间的分布式事务，保证数据一致性
三、实验方案
rabbitmq作为消息中间件
订单中心和运单中心分别作为消息的生产者和消息的消费者，通过rabbitmq传递消息
订单中心作为生产者，模拟用户创建订单，在本地持久化订单信息，记录消息的状态信息，并将消息发送到rabbitmq，同时开启confirm机制，接收消息中间件rabbitmq的响应信息，更新本地消息发送状态（定时任务轮训消息状态信息表，一定时间内未发送成功的数据将再次发起推送，保证atlest
once.
运单中心作为消费者，消费rabbitmq中的订单信息，开启ack确认机制，确保不遗漏订单。并通过消息全局唯一ID保证数据的唯一性，不重复处理订单。
四、实验步骤
1、消息队列
1.1 rabbitmq安装过程略过。。。。
1.2 创建订单交换器：orderExchange
1.3 创建订单队列：orderQueue
1.4 绑定
2、数据库准备
2.1订单表
2.2 消息发送状态表
2.3 运单表
3、订单中心
3.1 订单中心分析
利用Rabbitmq发布确认机制(confirm)，确保发送成功的数据能被通知到
做个定时任务轮训发送失败以及发送后未响应的订单信息，重新发送
3.2 编写代码
3.2.1 Springboot整合rabbitmq和mysql数据库
3.2.1.1依赖如下：
<dependency>
<groupId></groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--rabbitmq-->
<dependency>
<groupId></groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--mysql-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<!--lombok-->
<dependency>
<groupId></groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!--jdbc-->
<dependency>
<groupId></groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<!--fastjson-->
<dependency>
<groupId></groupId>
<artifactId>fastjson</artifactId>
<version>1.2.17</version>
</dependency>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
3.2.1.2配置文件内容：
server:
port: 8080
spring:
datasource:
driver-class-name:
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
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
3.2.2 订单中心代码
import ;
import .slf4j.Slf4j;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
/**
* @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代表此方法是否发送成功
(new (){
@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 = (sql,());
if(count != 1){
("本地消息表状态修改失败");
}
}catch (Exception e){
("本息消息表状态修改异常",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作用：当收到消息回执时会带上这个参数
("orderExchange","",(),new CorrelationData((String) ("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 = (sql,("orderid"),("userid"),("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 = (sql,("orderid"),());
if(count != 1){
throw new Exception("记录消息发送状态失败");
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
3.3 订单中心测试
3.3.1 测试代码
@Autowired
private OrderService orderService;
@Test
public void orderServiceTest() throws Exception {
//生成订单信息
JSONObject orderinfo = new JSONObject();
("orderid",().toString());
("userid",().toString());
("goodsid",().toString());
(orderinfo);
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
3.3.2 测试验证结果
orderQueue消息队列中已经接收到数据
订单表里的数据
状态表数据：
4、运单中心
4.1 运单中心分析
消费者收到消息进行处理，处理成功则发送ACK消息通知MQ清除该条记录，否则通知MQ重发或者等待MQ自动重发。本地维护一个处理次数，如果多次处理仍然失败，则将该消息丢弃或者加入到死信队列（DLQ）中。死信队列中的数据可以人工干预。
4.2 编写代码
import ;
import ;
import .slf4j.Slf4j;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
/**
* @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 = (message);
("收到MQ里面的消息：" + ());
(1000L);
//执行业务操作，同一个数据不能处理两次，根据业务情况去重，保证幂等性
String orderid = ("orderid");
//分配快递员配送
dispatch(orderid);
//ack 通知MQ数据已经收到
(tag,false);
}catch (Exception e){
//异常情况，需要根据需求去重发或者丢弃
//重发一定次数后丢弃，日志告警（rabbitmq没有设置重发次数功能，重发时需要代码实现，比如使用redis记录重发次数，）
(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 = (sql,orderid,"东哥","配送中");
if(count != 1){
throw new Exception("调度数据插入失败，原因[数据库操作]");
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
4.3 订单中心测试
启动springboot后自动监听MQ中的消息队列，自动处理
测试结果如下：