分布式事务项目实战

左手的ㄟ右手 2022-04-24 00:44 289阅读 0赞

网上找了个小项目,然后觉得收获很大,有必要写出来分享hhh

# **自动选课需求** #

1.支付成功即完成订单，订单完成之后系统需自动添加选课。

2.下图是微信支付、学成在线订单服务、学成在线学习服务交互图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70][]

1、用户支付完成，微信支付系统会主动通知学成在线支付结果，学成在线也可主动请求微信支付查询订单的支付结果。最终得到支付结果后将订单支付结果保存到订单数据库中。

2、订单支付完成系统自动向选课表添加学生选课记录。

3、选课记录添加完成学习即可在线开始学习。

### **订单数据库** ###

**订单任务表**     --订单付款后,往订单任务表插入一条数据,

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1][]

在任务表中包括了交换机的名称、路由key等信息为了是将任务的处理做成一个通用的功能。考虑分布式系统并发读取任务处理任务的情况发生项目使用乐观锁的方式解决并发问题

**已完成任务表**： 自动给用户添加完成选课后,往该表插入一条记录,并删除任务表的记录

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2][]

### **选课数据库**  ###

**学生选课表  ->  ** 订单完成后往该表插入一条记录

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3][]

**学生选课记录表    ->** 历史任务表,选课后往历史记录表插入一条记录

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 4][]

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

由于订单付款成功,那么该课程自动添加进选课,他们直接的关系是付款成功那么一定要把课程进行选课'

该项目综合考虑选择基于消息的分布式事务解决方案，解决方案如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 5][]

1. 订单完成后,在订单任务表插入一条信息,

2.使用定时器按时扫描订单任务表,并使用mq发送队列消息

3.学习服务接收消息队列消息,把信息插入选课表后,再插入一条历史记录表,然后通过mq响应结果

4.订单服务接收到mq后,把任务表的数据删除,然后再往历史记录表中插入数据,更新选课事件为完成

异常分析

如果订单发送mq出现了异常,那么任务表中的任务会一直存在,会一直发送,知道成功了为止.该节点奔溃不会影响数据结构

如果课程中心选课异常,那么将无法发送消息到订单中心标记完成任务,订单中心依旧会发送消息给课程中心,直到成功了为止,(需要注意的是,课程选课需要实现幂等性)

该思路是没有问题的,不会出现数据丢失的情况

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

### **Spring Task定时任务** ###

根据分布式事务的研究结果，订单服务需要定时扫描任务表向MQ发送任务。本节研究定时任务处理的方案，并实  
现定时任务扫描任务表并向MQ发送消息.

实现定时任务的方案如下：

1、使用jdk的Timer和TimerTask实现:可以实现简单的间隔执行任务，无法实现按日历去调度执行任务。

2、使用Quartz实现Quartz 是一个异步任务调度框架，功能丰富，可以实现按日历调度。

3、使用Spring Task实现Spring 3.0后提供Spring Task实现任务调度，支持按日历调度，相比Quartz功能稍简单，但是在开发基本够用，支持注解编程方式。

选择spring Task,别问我为什么,spring忠实粉丝,哈哈哈

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

# **订单服务定时发送消息** #

定时任务发送消息流程如下：

1、每隔1分钟扫描一次任务表。一次取出多个任务，取出超过1分钟未处理的任务

2、考虑订单服务可能集群部署，为避免重复发送任务使用乐观锁的方式每次从任务列表取出要处理的任务3、任务发送完毕更新任务发送时间

RabbitMQ配置

声明订单交换机,绑定俩个队列,并设置路由规则

package com.xuecheng.order.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMQConfig {
        //添加选课任务交换机
        public static final String EX_LEARNING_ADDCHOOSECOURSE = "ex_learning_addchoosecourse";
    
        //完成添加选课消息队列
        public static final String XC_LEARNING_FINISHADDCHOOSECOURSE = "xc_learning_finishaddchoosecourse";
    
        //添加选课消息队列
        public static final String XC_LEARNING_ADDCHOOSECOURSE = "xc_learning_addchoosecourse";
    
    
        //添加选课路由key
        public static final String XC_LEARNING_ADDCHOOSECOURSE_KEY = "addchoosecourse";
        //完成添加选课路由key
        public static final String XC_LEARNING_FINISHADDCHOOSECOURSE_KEY = "finishaddchoosecourse";
    
        /**
         * 交换机配置
         * @return the exchange
         */
        @Bean(EX_LEARNING_ADDCHOOSECOURSE)   //订单交换机
        public Exchange EX_DECLARE() {
            return ExchangeBuilder.directExchange(EX_LEARNING_ADDCHOOSECOURSE).durable(true).build();
        }
        //声明队列
        @Bean(XC_LEARNING_FINISHADDCHOOSECOURSE)  //完成选课队列
        public Queue QUEUE_finishaddchoose() {
            Queue queue = new Queue(XC_LEARNING_FINISHADDCHOOSECOURSE,true,false,true);
            return queue;
        }
    
    
        @Bean(XC_LEARNING_ADDCHOOSECOURSE)  //完成选课队列
        public Queue QUEUE_addchoose() {
            Queue queue = new Queue(XC_LEARNING_ADDCHOOSECOURSE,true,false,true);
            return queue;
        }
        /**
         * 绑定队列到交换机 .
         * @param queue    the queue
         * @param exchange the exchange
         * @return the binding
         */
        @Bean    //完成选课队列绑定交换机
        public Binding BINDING_QUEUE_FINISHADDCHOOSECOURSE(@Qualifier(XC_LEARNING_FINISHADDCHOOSECOURSE) Queue queue, @Qualifier(EX_LEARNING_ADDCHOOSECOURSE) Exchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(XC_LEARNING_FINISHADDCHOOSECOURSE_KEY).noargs();
        }
    
        @Bean    //添加选课队列绑定交换机
        public Binding BINDING_QUEUE_ADDCHOOSECOURSE(@Qualifier(XC_LEARNING_ADDCHOOSECOURSE) Queue queue, @Qualifier(EX_LEARNING_ADDCHOOSECOURSE) Exchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(XC_LEARNING_ADDCHOOSECOURSE_KEY).noargs();
        }
    
    }

***Repository***

在XcTaskRepository中自定义方法如下： 使用的是jpa,查询发布一分钟后的任务,,,

import com.github.pagehelper.Page;
    import com.xuecheng.framework.domain.task.XcTask;
    import org.springframework.data.domain.Pageable;
    import org.springframework.data.jpa.repository.JpaRepository;
    
    import java.util.Date;
    
    public interface XcTaskRepository extends JpaRepository<XcTask,String> {
    
    //    查询1分钟前的任务
        Page<XcTask> findByUpdateTimeBefore(Pageable pageable, Date updateTime);
    
       @Modifying  //使用乐观锁,放着多个服务器抢到同一个任务
        @Query("update XcTask t set t.updateTime = :updateTime where t.id = :id ")
        public int updateTaskTime(@Param(value = "id") String id,@Param(value =        "updateTime")Date updateTime);
    }

**Service**

import com.github.pagehelper.Page;
    import com.xuecheng.framework.domain.task.XcTask;
    import com.xuecheng.framework.model.response.CommonCode;
    import com.xuecheng.framework.model.response.ResponseResult;
    import com.xuecheng.order.dao.XcTaskRepository;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.domain.PageRequest;
    import org.springframework.data.domain.Pageable;
    import org.springframework.stereotype.Service;
    
    import java.util.Date;
    import java.util.List;
    import java.util.Optional;
    
    @Service
    public class TaskService {
    
        @Autowired
        XcTaskRepository xcTaskRepository;
    
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
    //    取出前n条记录
        public List<XcTask> findTaskList(int n, Date updateTime){
    
            //设置分页参数，取出前n 条记录
            Pageable pageable = new PageRequest(0, n);
    
            Page<XcTask> xcTasks = xcTaskRepository.findByUpdateTimeBefore(pageable, updateTime);
            return xcTasks.getResult();
        }
    
    
        public void publish(XcTask xcTask,String ex,String routingKey){
    
            Optional<XcTask> optional = xcTaskRepository.findById(xcTask.getId());//效验任务是否存在
    
            if(optional.isPresent()){
                XcTask one = optional.get();
                rabbitTemplate.convertAndSend(ex,routingKey ,one);
    
                //更新任务时间为当前时间
                one.setUpdateTime(new Date());
                xcTaskRepository.save(one);
            }
    
    
        }
    }

发送选课消息 任务类

rabbitmq

package com.xuecheng.order.scheduling;
    
    import com.xuecheng.framework.domain.task.XcTask;
    import com.xuecheng.order.config.RabbitMQConfig;
    import com.xuecheng.order.service.TaskService;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.Calendar;
    import java.util.Date;
    import java.util.GregorianCalendar;
    import java.util.List;
    
    @Component
    public class ChooseCourseTask {
    
        @Resource
        TaskService taskService;
    
    
        @Scheduled(fixedDelay = 6000)
        public void sendChoosecourseTask(){
    
    //        取出当前时间一分钟钱的时间
            Calendar calendar = new GregorianCalendar();
            calendar.setTime(new Date());
            calendar.set(GregorianCalendar.MINUTE, -1);  //前一分钟的记录
    
            Date time = calendar.getTime();
    
            List<XcTask> taskList = taskService.findTaskList(1000, time);
    
    
            //发送任务
            for (XcTask xcTask : taskList) {
                taskService.publish(xcTask, RabbitMQConfig.EX_LEARNING_ADDCHOOSECOURSE, RabbitMQConfig.XC_LEARNING_ADDCHOOSECOURSE_KEY);
            }
        }
    }

考虑订单服务将来会集群部署，为了避免任务在1分钟内重复执行，这里使用乐观锁，实现思路如下：  
1) 每次取任务时判断当前版本及任务id是否匹配，如果匹配则执行任务，如果不匹配则取消执行。  
2) 如果当前版本和任务Id可以匹配到任务则更新当前版本加1

1、在Dao中增加校验当前版本及任务id的匹配方法

//使用乐观锁方式校验任务id和版本号是否匹配，匹配则版本号加1
    @Modifying@Query("update XcTask t set t.version = :version+1 where t.id = :id and t.version =:version")
    public int updateTaskVersion(@Param(value = "id") String id,@Param(value = "version") int version);

2.在service中增加方法，使用乐观锁方法校验任务

public int  getTask(XcTask task){
            return xcTaskRepository.updateTaskVersion(task.getId(), task.getVersion());
        }

3、执行任务类中修改

package com.xuecheng.order.scheduling;
    
    import com.xuecheng.framework.domain.task.XcTask;
    import com.xuecheng.order.config.RabbitMQConfig;
    import com.xuecheng.order.service.TaskService;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.Calendar;
    import java.util.Date;
    import java.util.GregorianCalendar;
    import java.util.List;
    
    @Component
    public class ChooseCourseTask {
    
        @Resource
        TaskService taskService;
    
    
        @Scheduled(fixedDelay = 6000)
        public void sendChoosecourseTask(){
    
    //        取出当前时间一分钟钱的时间
            Calendar calendar = new GregorianCalendar();
            calendar.setTime(new Date());
            calendar.set(GregorianCalendar.MINUTE, -1);  //前一分钟的记录
    
            Date time = calendar.getTime();
    
            List<XcTask> taskList = taskService.findTaskList(1000, time);
    
    
            //发送任务
            for (XcTask xcTask : taskList) {
    
    //          使用乐观锁,确认是否发送信息
                if (taskService.getTask(xcTask) > 0){
                    //   数据库中的 version  跟   task传入的version一致,才能被更新,
                    taskService.publish(xcTask, RabbitMQConfig.EX_LEARNING_ADDCHOOSECOURSE, RabbitMQConfig.XC_LEARNING_ADDCHOOSECOURSE_KEY);
                }
            }
        }
    }

# **自动选课课程开发** #

需求分析

学习服务接收MQ发送添加选课消息，执行添加 选 课操作。  
添加选课成功向学生选课表插入记录、向历史任务表插入记录、并向MQ发送“完成选课”消息。

**Dao**

@Repository
    public interface XcLearningCourseRepository extends JpaRepository<XcLearningCourse,String> {
        //根据用户和课程查询选课记录，用于判断是否添加选课
        XcLearningCourse findXcLearningCourseByUserIdAndCourseId(String userId, String courseId);
    
    
    }
    
    
    
    @Repository//历史任务dao
    public interface XcTaskHisRepository extends JpaRepository<XcTaskHis,String> {
    }

**Service**

1、添加选课方法

向xc\_learning\_course添加记录，为保证不重复添加选课，先查询历史任务表，如果从历史任务表查询不到任务说明此任务还没有处理，此时则添加选课并添加历史任务。

package com.xuecheng.learning.service;
    
    import com.xuecheng.framework.domain.learning.XcLearningCourse;
    import com.xuecheng.framework.domain.task.XcTask;
    import com.xuecheng.framework.domain.task.XcTaskHis;
    import com.xuecheng.framework.model.response.CommonCode;
    import com.xuecheng.framework.model.response.ResponseResult;
    import com.xuecheng.learning.dao.XcLearningCourseRepository;
    import com.xuecheng.learning.dao.XcTaskHisRepository;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.beans.BeanUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.Date;
    
    //import com.xuecheng.framework.domain.learning.respones.GetMediaResult;
    //import com.xuecheng.framework.domain.learning.respones.LearningCode;
    
    /**
     * @author Administrator
     * @version 1.0
     **/
    @Service
    public class LearningService {
    
        @Autowired
        XcTaskHisRepository xcTaskHisRepository;
    
        @Autowired
        XcLearningCourseRepository xcLearningCourseRepository;
    
        @Transactional
        public ResponseResult addCourse(String userId, String courseId, String valid, Date
                startTime, Date endTime, XcTask xcTask){
    
            if (StringUtils.isBlank(userId)  ||StringUtils.isBlank(courseId) ){
    //            异常处理
                return new ResponseResult(CommonCode.FAIL);
            }
    
            if(xcTask == null || StringUtils.isEmpty(xcTask.getId())){
                return new ResponseResult(CommonCode.FAIL);   //假装有很认真处理
            }
    
            //效验任务是否存在
            XcTaskHis one = xcTaskHisRepository.getOne(xcTask.getId());
    
            if (one != null){
                //已经执行,无需执行
                return new ResponseResult(CommonCode.SUCCESS);
            }
    
            XcLearningCourse xcLearningCourse = xcLearningCourseRepository.findXcLearningCourseByUserIdAndCourseId(userId, courseId);//效验用户是否插入过该课程
    
    
            if (xcLearningCourse != null){//选课记录存在,修改状态及基本信息
                xcLearningCourse.setValid(valid);
                xcLearningCourse.setStartTime(startTime);
                xcLearningCourse.setEndTime(endTime);
                xcLearningCourse.setStatus("501001");
                xcLearningCourseRepository.save(xcLearningCourse);
            }else {
                //插入课程记录
    
                xcLearningCourse = new XcLearningCourse();
                xcLearningCourse.setUserId(userId);
                xcLearningCourse.setCourseId(courseId);
                xcLearningCourse.setValid(valid);
                xcLearningCourse.setStartTime(startTime);
                xcLearningCourse.setEndTime(endTime);
                xcLearningCourse.setStatus("501001");
                xcLearningCourseRepository.save(xcLearningCourse);
            }
    
    
    //        像历史记录表插入记录
            XcTaskHis xcTaskHis = new XcTaskHis();
    
            BeanUtils.copyProperties(xcTask, xcTaskHis);
            xcTaskHisRepository.save(xcTaskHis);
            return new ResponseResult(CommonCode.SUCCESS);
        }
    
    }

**接收添加选课消息**

接收到添加选课的消息调用添加选课方法完成添加选课，并发送完成选课消息。

import com.alibaba.fastjson.JSON;
    import com.xuecheng.framework.domain.task.XcTask;
    import com.xuecheng.framework.model.response.ResponseResult;
    import com.xuecheng.learning.config.RabbitMQConfig;
    import com.xuecheng.learning.service.LearningService;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.Map;
    
    @Component
    public class ChooseCourseTask {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(ChooseCourseTask.class);
    
        @Autowired
        LearningService learningService;
        @Autowired
        RabbitTemplate rabbitTemplate;
    
    
        @RabbitListener(queues = {RabbitMQConfig.XC_LEARNING_ADDCHOOSECOURSE})
        public void receiveChoosecourseTask(XcTask xcTask){
    
            //接收到 的消息id
            String id = xcTask.getId();
    
            try {
                //参数封装,转换
                String requestBody = xcTask.getRequestBody();//JSON数据,转换成Map
                Map map = JSON.parseObject(requestBody, Map.class);
                String userId = (String) map.get("userId");
                String courseId = (String) map.get("courseId");
                String valid = (String) map.get("valid");
    
                Date startTime =  null;
                Date endTime = null;
    
                SimpleDateFormat dateFormat = new SimpleDateFormat("YYYY-MM-dd HH:mm:dd");
                if(map.get("startTime")!=null){
                    startTime = dateFormat.parse((String) map.get("startTime"));
                }
                if(map.get("endTime")!=null){
                    endTime =dateFormat.parse((String) map.get("endTime"));
                }
    
    
                //添加选课
                ResponseResult addCourse = learningService.addCourse(userId, courseId, null, startTime, endTime, xcTask);
    
    
                if (addCourse.isSuccess()){
                    //执行成功.....
    
    //                发送mq到订单中心,响应完成
                    rabbitTemplate.convertAndSend(RabbitMQConfig.EX_LEARNING_ADDCHOOSECOURSE, RabbitMQConfig.XC_LEARNING_FINISHADDCHOOSECOURSE_KEY, xcTask);
                }
            }catch (Exception e){
                LOGGER.error("send finish choose course taskId:{}", id);
            }
        }
    }

**订单服务结束任务**

订单服务接收MQ完成选课的消息，将任务从当前任务表删除，将完成的任务添加到完成任务表。

service

//删除任务
        @Transactional
        public void finishTask(String taskId){
            Optional<XcTask> taskOptional = xcTaskRepository.findById(taskId);
    
    
            if (taskOptional.isPresent()){
                XcTask xcTask = taskOptional.get();
                //任务历史记录
                XcTaskHis xcTaskHis = new XcTaskHis();
                BeanUtils.copyProperties(xcTask, xcTaskHis);
    //            保存历史记录
    
                xcTaskHlsRepository.save(xcTaskHis);
                xcTaskRepository.deleteById(xcTask.getId());
            }
    
    }

最后用mq 监听一下,调用方法就行啦

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70]: /images/20220221/4a016f2c6f134de1806851c6d72d5046.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1]: /images/20220221/4e3945c014844bdeba7f3a4f898f11cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2]: /images/20220221/c35a443017d84eefb4c7de143b7a2c58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3]: /images/20220221/d61cf9e4a4e345b6bf29b6601b19af82.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 4]: /images/20220221/163c6c56e36747419eb8d2d5273d81dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 5]: /images/20220221/0385e5f9e42947328e6785bf1e6d0aed.png