【103期】RabbitMQ 实现多系统间的分布式事务，保证数据一致性|mysql|param|sql

点击上方“Java精选”，选择“设为星标”

别问别人为什么，多问自己凭什么！

下方留言必回，有问必答！

每天08:08更新文章，每天进步一点点...

一、实验环境

Lunix系统：Centos7.5
安装软件：rabbitmq
开发工具：IDEA

二、实验目的

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

三、实验方案

rabbitmq作为消息中间件

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

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

3.2 编写代码

3.2.1 Springboot整合rabbitmq和mysql数据库

3.2.1.1依赖如下：

org.springframework.bootgroupId> spring-boot-starter-webartifactId> dependency>

org.springframework.bootgroupId> spring-boot-starter-amqpartifactId> dependency>

mysqlgroupId> mysql-connector-javaartifactId> runtimescope> dependency>

org.projectlombokgroupId> lombokartifactId> trueoptional> dependency>

org.springframework.bootgroupId> spring-boot-starter-jdbcartifactId> 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: MANUAL3.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消息队列中已经接收到数据

订单表里的数据

状态表数据：

4、运单中心 4.1 运单中心分析

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中的消息队列，自动处理

测试结果如下：