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

Java技术精选

共 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
1.3 创建订单队列:orderQueue
1.4 绑定

2、数据库准备

2.1订单表
2.2 消息发送状态表
2.3 运单表

3、订单中心

3.1 订单中心分析

利用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消息队列中已经接收到数据

订单表里的数据

状态表数据:

4、运单中心

4.1 运单中心分析

消费者收到消息进行处理,处理成功则发送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中的消息队列,自动处理

测试结果如下:

感谢阅读,希望对你有所帮助 :) 

来源:blog.csdn.net/qq_31463999/article/details/79220061



浏览 41
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报