MQ实现分布式事务

缺乏、安全感 2022-03-30 03:22 270阅读 0赞

**目录**

detail

项目地址：

总结

--------------------

分布式事务不像单体应用一样使用@transactional进行事务管理即可

分布式事务不能保证强一致性，只能保证最终一致性，我使用mq来实现不同系统之间的通讯，但是mq也会出现错误的时候，还有不方便保存mq消息，所以再做一个消息服务来储存、定时发送消息，失败消费消息时进行重发消息，记录死亡消息等等。

这个项目让a系统减少一，b系统也减少一，activitymq作为通讯

a，b系统基本的操作数据库，不做解释，下面讲下消息服务：

# detail #

sql：

CREATE TABLE `message`  (
      `id` varchar(36) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
      `message` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '消息内容\r\n',
      `sendcount` bigint(20) NULL DEFAULT NULL COMMENT '重复发送消息次数',
      `queue` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '队列名称',
      `sendsystem` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '发送消息的系统',
      `status` int(255) NULL DEFAULT NULL COMMENT '状态：0等待消费 1已消费 2 已经死亡',
      `customerdate` timestamp(0) NULL DEFAULT NULL COMMENT '消费时间',
      `customersystem` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '消费消息的系统',
      `cdate` timestamp(0) NULL DEFAULT NULL COMMENT '创建时间',
      `diecount` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '死亡次数',
      `diedate` datetime(0) NULL DEFAULT NULL COMMENT '死亡时间',
      PRIMARY KEY (`id`) USING BTREE,
      INDEX `id`(`id`) USING BTREE
    ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;
    
    SET FOREIGN_KEY_CHECKS = 1;

service类

package com.example.demo.service;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.jdbc.core.PreparedStatementSetter;
    import org.springframework.stereotype.Service;
    
    import java.sql.PreparedStatement;
    import java.sql.SQLException;
    import java.util.List;
    
    /**
     * @program: demoA
     * @description
     * @author: dajitui
     * @create: 2019-01-03 02:20
     **/
    @Service
    public class messageService {
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
        public List getMessage(){
            return jdbcTemplate.queryForList("SELECT * FROM message WHERE STATUS = 0 AND sendcount<=10");
        }
    
        public int updateMessage(String id){
            //SQL+结果`
            int resRow = jdbcTemplate.update("UPDATE message SET sendcount=sendcount+1 WHERE id=?", new PreparedStatementSetter() {
                //映射
                // 数据
                @Override
                public void setValues(PreparedStatement ps) throws SQLException {
                    ps.setString(1, id);
                }
            });
            //返回结果
            return resRow;
        }
    
        public int updateMessageStatusAndDate(){
            //SQL+结果`
            int resRow = jdbcTemplate.update("UPDATE message SET status = 2 , diedate = NOW() where sendcount=10 ", new PreparedStatementSetter() {
                //映射
                // 数据
                @Override
                public void setValues(PreparedStatement ps) throws SQLException {
                    //ps.setString(1, id);
                }
            });
            //返回结果
            return resRow;
        }
    
        public int updateMessageCustomerDate(String id){
            //SQL+结果`
            int resRow = jdbcTemplate.update("UPDATE message SET customerdate=NOW() , STATUS=1 WHERE id = ?", new PreparedStatementSetter() {
                //映射
                // 数据
                @Override
                public void setValues(PreparedStatement ps) throws SQLException {
                    ps.setString(1, id);
                }
            });
            //返回结果
            return resRow;
        }
    }

Controller,为了使用fegin进行访问

package com.example.demo.controller;
    
    import com.example.demo.service.messageService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.jdbc.core.PreparedStatementSetter;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.sql.PreparedStatement;
    import java.sql.SQLException;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.UUID;
    
    /**
     * @program: demoA
     * @description
     * @author: dajitui
     * @create: 2018-12-31 17:04
     **/
    @RestController
    public class bController {
    
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
        @Autowired
        private messageService messageService;
    
        @RequestMapping(value = "/message")
        public int b(@RequestParam(value = "message")String message,@RequestParam(value = "queue")String queue,@RequestParam(value = "sendsystem")String sendsystem,@RequestParam(value = "customersystem")String customersystem){
    
                String id= UUID.randomUUID().toString();
    
                String sendcount="0";
    
                //等待消费
                String status="0";
    
                //创建时间
            //创建时间
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
            String date=format.format(new Date());
    
                String diecount="0";
    
                //SQL+结果
                int resRow = jdbcTemplate.update("INSERT INTO `message`(`id`, `message`, `sendcount`, `queue`, `sendsystem`, `status`, `customersystem`, `cdate`, `diecount`) VALUES (?, ?, ?, ?, ?, ?, ?, ?,?);", new PreparedStatementSetter() {
                    //映射
                    // 数据
                    @Override
                    public void setValues(PreparedStatement ps) throws SQLException {
                        ps.setString(1, id);
                        ps.setString(2,message);
                        ps.setString(3,sendcount);
                        ps.setString(4,queue);
                        ps.setString(5,sendsystem);
                        ps.setString(6,status);
                        ps.setString(7,customersystem);
                        ps.setString(8, date);
                        ps.setString(9,diecount);
    
                    }
                });
                //返回结果
                return resRow;
    
        }
    
        @RequestMapping(value = "/message/update")
        public int updatemessage(@RequestParam(value = "id")String id){
            return messageService.updateMessage(id);
        }
    
        @RequestMapping(value = "/message/custom")
        public int updatemessagedate(@RequestParam(value = "id")String id){
            return messageService.updateMessageCustomerDate(id);
        }
    }

定时任务进行查看未消费且重复次数不超过10的消息，还有如果超过10次则设置死亡时间，消息状态

package com.example.demo.task;
    
    
    import com.example.demo.entity.message;
    import com.example.demo.mq.JMSProducer;
    import com.example.demo.service.messageService;
    import com.netflix.discovery.converters.Auto;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.jms.Destination;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    @Component
    public class ScheduledTasks {
    
        @Autowired
        private messageService messageService;
    
        @Autowired
        private JMSProducer jmsProducer;
    
        private static SimpleDateFormat dateFormat=new SimpleDateFormat("yyyy-MMM-ddd HH:mm:ss");
    
    
        /**
         * 定时查看未消费且重发次数少于10次的消息，并发送到mq
         */
        @Scheduled(fixedRate = 60000,initialDelay = 100)
        public void a(){
            System.out.println("现在时间："+dateFormat.format(new Date())+"开始查找未消费且重新发送次数不超过10次的消息......");
            List<Map<String,Object>> list=messageService.getMessage();
            if(list.size()!=0) {
                for (Map map : list) {
                    String queue = map.get("queue").toString();
                    String message = map.get("message").toString();
                    String id = map.get("id").toString();
                    Map me = new HashMap();
                    me.put("message", "\"" + message + "\"");
                    me.put("id", "\"" + id + "\"");
    
                    try {
                        Destination destination = new ActiveMQQueue(queue);
                        jmsProducer.sendMessage(destination, me.toString());
                        int result = messageService.updateMessage(id);
                        if (result == 0) {
                            System.out.println("更新message发送次数失败!");
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
    
                }
            }
        }
    
        /**
         * 定时查看未消费且重发次数已经等于10次的消息，修改消息状态以及死亡时间
         */
        @Scheduled(fixedRate = 60000,initialDelay = 5000)
        public void b(){
            System.out.println("现在时间："+dateFormat.format(new Date())+"开始查找并更新已死亡的消息......");
            messageService.updateMessageStatusAndDate();
        }
    }

在启动类加上

@EnableScheduling

将获取的消息队列，通过mq进行发送

application.properties

spring.activemq.broker-url=tcp://127.0.0.1:61616
    spring.activemq.user=admin
    spring.activemq.password=admin
    spring.activemq.pool.enabled=true
    spring.activemq.pool.max-connections=50
    spring.activemq.pool.expiry-timeout=10000
    spring.activemq.pool.idle-timeout=30000

生产消息

package com.example.demo.mq;
    
    import com.example.demo.service.messageService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.stereotype.Component;
    
    import javax.jms.Destination;
    
    
    /**
     * @program: demoA
     * @description
     * @author: dajitui
     * @create: 2019-01-03 14:17
     **/
    @Component
    public class JMSProducer {
        @Autowired
        private JmsTemplate jmsTemplate;
    
        @Autowired
        private messageService messageService;
    
        public void sendMessage(Destination destination, String message) {
            this.jmsTemplate.convertAndSend(destination,message);
        }
    }

消费消息

package com.example.demo.mq;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Component;
    
    /**
     * @program: demoA
     * @description
     * @author: dajitui
     * @create: 2019-01-03 14:18
     **/
    @Component
    public class JMSConsumer {
        private final static Logger logger = LoggerFactory.getLogger(JMSConsumer.class);
    
        @JmsListener(destination = "springboot.queue.test")
        public void receiveQueue(String msg) {
            logger.info("接收到消息：{}",msg);
        }
    }

# 项目地址： #

[https://github.com/dajitui/MQ-][https_github.com_dajitui_MQ-]

如果有多个消息服务并行进行，需要使用分布式锁进行控制[https://blog.csdn.net/weixin\_38336658/article/details/85786818][https_blog.csdn.net_weixin_38336658_article_details_85786818]

# 总结 #

像现在很多公司比较使用分布式事务框架去实现，应该维护开发代价都很大。像mq可以满足很多场景，主要实现：重试（补偿机制），幂等性处理！！！

TCC框架：ByteTcc,Hmilu

XA:LCN一般很少人用，缺陷很多

[https_github.com_dajitui_MQ-]: https://github.com/dajitui/MQ-
[https_blog.csdn.net_weixin_38336658_article_details_85786818]: https://blog.csdn.net/weixin_38336658/article/details/85786818