RabbitMQ实现分布式事务，保证数据一致性

一、实验环境

Lunix系统：Centos7.5

安装软件：rabbitmq

开发工具：IDEA

二、实验目的

Rabbitmq实现多系统间的分布式事务，保证数据一致性

三、实验方案

rabbitmq作为消息中间件

订单中心和运单中心分别作为消息的生产者和消息的消费者，通过rabbitmq传递消息

订单中心作为生产者，模拟用户创建订单，在本地持久化订单信息，记录消息的状态信息，并将消息发送到rabbitmq，同时开启confirm机制，接收消息中间件rabbitmq的响应信息，更新本地消息发送状态（定时任务轮训消息状态信息表，一定时间内未发送成功的数据将再次发起推送，保证atlestonce.

运单中心作为消费者，消费rabbitmq中的订单信息，开启ack确认机制，确保不遗漏订单。并通过消息全局唯一ID保证数据的唯一性，不重复处理订单。

四、实验步骤

1、消息队列

1.1 rabbitmq安装过程略过。。。。
1.2 创建订单交换器：orderExchange

1.3 创建订单队列：orderQueue

1.4 绑定

2、数据库准备

2.2 消息发送状态表

2.3 运单表

3、订单中心

3.1 订单中心分析

利用Rabbitmq发布确认机制(confirm)，确保发送成功的数据能被通知到做个定时任务轮训发送失败以及发送后未响应的订单信息，重新发送。

3.2 编写代码

3.2.1 Springboot整合rabbitmq和mysql数据库

3.2.1.1依赖如下：


<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--rabbitmq-->
<dependency><groupId>org.springframework.boot</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>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional>
</dependency>
<!--jdbc-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<!--fastjson-->
<dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.17</version>
</dependency>

3.2.1.2配置文件内容：

server:port: 8080spring:datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8&allowMultiQueries=trueusername: rootpassword: root123rabbitmq:host: localhostport: 5672username: adminpassword: admin123virtual-host: /#必须配置这个，生产者才会确认回调publisher-confirm-type: correlatedpublisher-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;/*** 订单中心*/
@Slf4j
@Service
public class OrderService {@Autowiredprivate JdbcTemplate jdbcTemplate;@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void setup(){//消息发送完成后，则回调此方法，ack代表此方法是否发送成功rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){@Overridepublic 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*/@Transientprivate 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消息队列中已经接收到数据

订单表里的数据

状态表数据：

4、运单中心

4.1 运单中心分析

消费者收到消息进行处理，处理成功则发送ACK消息通知MQ清除该条记录，否则通知MQ重发或者等待MQ自动重发。本地维护一个处理次数，如果多次处理仍然失败，则将该消息丢弃或者加入到死信队列（DLQ）中。死信队列中的数据可以人工干预。

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;/*** 运单系统*/
@Slf4j
@Service
public class DispatchService {@Autowiredprivate 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 订单编号*/@Transientprivate 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中的消息队列，自动处理

测试结果如下：