分布式事务讲解 -消息队列+定时任务+本地事件表

怼烎@ 2023-09-29 15:21 23阅读 0赞

### 文章目录 ###

*  1:使用场景介绍
 *  2：方案原理介绍
 *  3：开发步骤-生产者
 *   *  1：pom
     *  2：application配置
     *  3：配置mq
     *  4：配置定时任务，完成支付状态发入mq
     *  5：开启启动类注解
 *  4：开发步骤-消费者
 *   *  1-2 基本配置一样
     *  3：配置mq，包括重试机制
     *  4：配置消费者消费

# 1:使用场景介绍 #

这种方案是基于分布式服务理论从业务层面做出的设计，【消息队列+定时任务+本地事件表】方案就是消息队列搭配上本地事件表实现的分布式事务，这种方案的特点就是通过服务和消息队列的整合，可以对服务实现解耦，缺点是异步执行  
比较适用于哪些业务简单，数据量小，并且不要求同步执行的场景。说白了，如果项目选型时，大部分人不会分布式事务框架，可以用这样的业务设计来提到

# 2：方案原理介绍 #

还是按照在电商系统中，调用支付系统完成后，然后支付系统调用订单系统更改订单状态

![在这里插入图片描述][5a1ca5222a90431aa1ebd6966f84e758.png]  
注：

> 在2-1，-2，-3这样的场景中，一般是先调用本地db，然后再调用第三方。  
> 其实其他业务也一样，比如我们假设一个场景，如果我们需要调用第三方接口，还要执行我们自己的sql逻辑，那么开发时的顺序肯定是先执行调用第三方的，然后再执行我们的。毕竟我们自己的db可以通过事务回滚，第三方已经调走了，收不回来了；

# 3：开发步骤-生产者 #

## 1：pom ##

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.3.2.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.mashibing.serviceorder</groupId>
        <artifactId>service-order</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>service-order</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
    
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-activemq</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-pool</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            
            <dependency>
                <groupId>org.mybatis.spring.boot</groupId>
                <artifactId>mybatis-spring-boot-starter</artifactId>
                <version>2.0.0</version>
            </dependency>
    
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid</artifactId>
                <version>1.1.12</version>
            </dependency>
    
            
            <dependency>
                <groupId>net.sf.json-lib</groupId>
                <artifactId>json-lib</artifactId>
                <version>2.4</version>
                <classifier>jdk15</classifier>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

## 2：application配置 ##

server:
      port: 5001
    
    #应用名称及验证账号
    spring:
      application:
        name: service-order
      activemq:
        broker-url: tcp://127.0.0.1:61616
        user: admin
        password: admin
        pool:
          enabled: true
          max-connections: 100
    
      datasource:
        type: com.alibaba.druid.pool.DruidDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        url: jdbc:mysql://localhost:3306/service-order?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
        username: root
        password: root
        dbcp2:
          initial-size: 5
          min-idle: 5
          max-total: 5
          max-wait-millis: 200
          validation-query: SELECT 1
          test-while-idle: true
          test-on-borrow: false
          test-on-return: false
    
    mybatis:
      mapper-locations:
      - classpath:mapper/*.xml

## 3：配置mq ##

// 配置消息发送端
    @Configuration
    public class ActiveMQConfig {
        
        @Value("${mqurl}")
        private String brokerUrl;
        // 创建队列，用于使用JMS发送消息
        @Bean
        public Queue queue() {
        
            return new ActiveMQQueue("ActiveMQQueue");
        }
        // 作为消息的发起方，创建一个消息队列连接工厂，供消费者连接和监听
        @Bean
        public ActiveMQConnectionFactory connectionFactory(){
        
            return new ActiveMQConnectionFactory(brokerUrl);
        }
    }

## 4：配置定时任务，完成支付状态发入mq ##

@Component
    public class ProduceTask {
        
        @Autowired
        private TblOrderEventDao tblOrderEventDao;
        @Autowired
        private Queue queue;
        @Autowired
        JmsMessagingTemplate jmsMessagingTemplate;
    	// 创建一分钟一次的定时任务
        @Scheduled(cron="0 */1 * * * ?")
        @Transactional(rollbackFor = Exception.class)
        public void task(){
        
        	// 取到数据库本地事务表状态为【新建】的数据
            List<TblOrderEvent> tblOrderEventList = tblOrderEventDao.selectByOrderType("0");
            for (int i = 0; i < tblOrderEventList.size(); i++) {
        
                TblOrderEvent event = tblOrderEventList.get(i);
                // 更改这条数据的状态为【已发送】
                tblOrderEventDao.updateEvent(event.getOrderType());
                System.out.println("修改数据库完成");
    			// 使用jmsMessagingTemplate发送信息至消息队列
                jmsMessagingTemplate.convertAndSend(	
                	queue,
                	JSONObject.fromObject(event).toString()
                );
            }
        }
    }

## 5：开启启动类注解 ##

@SpringBootApplication
    @EnableJms // 支持消息传递的注解
    @EnableScheduling   // 支持定时任务的注解
    public class ServiceOrderApplication {
        
        public static void main(String[] args) {
        
            SpringApplication.run(ServiceOrderApplication.class, args);
        }
    }

# 4：开发步骤-消费者 #

## 1-2 基本配置一样 ##

## 3：配置mq，包括重试机制 ##

// 配置消息消费端
    @Configuration
    public class ActiveMQConfig {
        
        @Value("${mqurl}")
        private String brokerUrl;
        @Value("${mquser}")
        private String user;
        @Value("${mqpassword}")
        private String password;
    
        /**
         * 作为消费方，根据activeMQ地址连接MQ工厂并设置重发机制
         * @param redeliveryPolicy
         * @return
         */
        @Bean
        public ActiveMQConnectionFactory connectionFactory
        (RedeliveryPolicy redeliveryPolicy){
        
            ActiveMQConnectionFactory activeMQConnectionFactory = 
            new ActiveMQConnectionFactory(user,password,brokerUrl);
            activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
            return activeMQConnectionFactory;
        }
        /**
         * 如果一个消息消费失败了，MQ工厂重发消息供消费者消费
         * @return
         */
        @Bean
        public RedeliveryPolicy redeliveryPolicy(){
        
            RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
            return redeliveryPolicy;
        }
        /**
         * 设置消息队列 确认机制
         * 设置手动确认，消费者确认之后才表明消费成功
         * @param activeMQConnectionFactory
         * @return
         */
        @Bean
        public JmsListenerContainerFactory jmsListenerContainerFactory
        (ActiveMQConnectionFactory activeMQConnectionFactory){
        
            DefaultJmsListenerContainerFactory bean = 
            new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(activeMQConnectionFactory);
            // 1: 自动确认，2： 客户端手动确认，3：自动批量确认，4 事务提交并确认。
            bean.setSessionAcknowledgeMode(2);
            return bean;
        }
    }

## 4：配置消费者消费 ##

@Component
    public class ConsumerQueue {
        
        @Autowired
        private TblPayEventDao tblPayEventDao;
        // destination属性值意为监听的目标，来源于消息发起者创建消息队列的nama属性值。
        // containerFactory属性值意为容器工厂，来源于消息消费者配置的监听消息的容器工厂。
        @JmsListener(destination = "ActiveMQQueue",
        	containerFactory = "jmsListenerContainerFactory")
        public void receive(TextMessage textMessage, Session session) 
        	throws JMSException {
        
            try {
        
                System.out.println("收到的消息："+textMessage.getText());
                String content = textMessage.getText();
    			// 将传来信息转成对象并操作数据库
                TblPayEvent tblPayEvent = 
                	(TblPayEvent) JSONObject.toBean(
                		JSONObject.fromObject(content), TblPayEvent.class);
                tblPayEventDao.insert(tblPayEvent);
                // 业务完成，确认消息 消费成功
                // 因为设置了手动确认，所以，需要调用消息的acknowledge方法
                textMessage.acknowledge();
            }catch (Exception e){
        
                // 回滚消息
                e.printStackTrace();
                System.out.println("异常了");
                // 消息消费失败，恢复回消息队列，等待重发再次消费。
                session.recover();
            }
    
        }
        /**
         * 监听死信队列，按需写方法内的业务逻辑
         * 补偿 处理（人工，脚本）。自己根据自己情况。
         * @param text
         */
        @JmsListener(destination = "DLQ.ActiveMQQueue")
        public void receive2(String text){
        
            //发邮件，人工干预
        }
    }

[5a1ca5222a90431aa1ebd6966f84e758.png]: https://img-blog.csdnimg.cn/5a1ca5222a90431aa1ebd6966f84e758.png