实现轻量级本地分布式事务

发布于:2021-07-18 08:50:50

简介

在上家公司时,由于机构 DIY 课程定制需要从固定课程复制,而复制需要调用三个小组的微服务,导致速度缓慢。最终通过id生成器,线程池,CompletionService ,闭锁实现 web 端调用的并发执行,提速优化同时保证三者之间事务安全, 接下来详细描述。


三个微服务分别为 课程创建、讲次创建、卷子创建,讲次挂在课程上,卷子挂在讲次上,关系如下:

?


主要的实现步骤为:


1、由id生成器提前生成好讲次id。


2、将生成好的讲次id分别传入 课程复制任务、讲次复制任务、讲次关联卷子复制任务。


3、由于最终需要存储课程id,所以课程创建结果id需要返回给主线程。


4、其中任意一个任务出错,全部回滚(回滚是通过调用各个微服务提供的清理方法实现)


?
?

?


CompleteService 源码解析

我们先来详细讲下关键类CompleteService,看看它在这里起到什么作用,以下对它的唯一实现类ExecutorCompletionService进行源码分析:


?
?

这里可以看到 ExecutorCompletionService 会将传入的线程池对象Executor作为它的成员对象,并且会创建一个名为completionQueue的LinkedBlockingQueue对象,这个对象的功能是关键,咱先接着往下看~


?
?

这里看到接收一个Callable类型的线程执行对象,并且执行


1、调用newTaskFor构造了RunnableFuture对象实际上构造了子类FutureTask,将Callable作为对象赋值给了它,并且初始化任务状态为NEW。
?

2、将RunnableFuture转换为FutureTask的子类QueueingFuture,并将它交给线程池执行。
?
这里看到了上面提到的completionQueue,在done()执行的时候,会把执行任务加入到completionQueue队列中。这其实就是ExecutorCompletionService能实现线程池每完成一个任务就是立马获取到执行结果的关键。


?
在task方法中,我们再次看到了熟悉的completionQueue,我们都知道Queue的take方法是一个阻塞获取结果的方法。 这时候我们设想一下,如果上面的done方法是每个任务完成时就执行的,那么通过take方法,我们将每次都能获取到最新的线程结果。 如果能想到这点,那么恭喜你,你已经基本掌握了CompleteService的用法了~ 那么接下来我们看下done是在什么时候执行的。


?

这里我们看到执行的两种结果,成功调用set方法,失败调用setException方法,分别看下set和setException的实现
?


set方法?可以看到执行成功后会将结果赋值给outcome对象,并会将 state 由 NEW 改为 NORMAL,接着调用finishCompletion方法。


可以看到finishCompletion最终调用done,将成功的任务放入Queue中。此时再通过take方法就可以获取FutureTask,并且通过FutureTask的get方法以上赋值的成功对象outcome。

?


setException方法?可以看到执行失败后会将异常对象赋值给outcome对象,并会将 state 由 NEW 改为 EXCEPTIONAL。 跟上面一样,接着执行finishCompletion方法,最终调用done,将失败的任务放入Queue中。此时再通过take方法获取FutureTask,并且通过FutureTask的get方法获取结果时,由于EXCEPTIONAL != NORMAL,此时将抛出异常对象outcome。

?


具体实现

流程图



?


实现步骤


1、由id生成器提前生成好讲次id。


2、将生成好的讲次id分别传入 课程复制任务、讲次复制任务、讲次关联卷子复制任务。


3、创建CountDownLatch对象,初始容量为2


4、创建线程池,线程数量为3。(由于线程池在这里还存在线程异常通讯的作用,因此每次请求新建,在这里未发挥出线程池真正的效果。在版本二中进行了优化)。


5、创建AtomicBoolean类型命名为classTypeCreateAlready的变量用于存储课程创建是否成功的结果,默认为false。


6、主线程实现描述:
主线程创建3个子线程后,将调用completeService的take方法阻塞,直到返回结果FutureTask:


get得到课程id结果,此时说明创建成功,设置classTypeCreateAlready为true,作为“讲次创建任务” 和 “卷子创建任务” 判断课程是否创建成功的依据。

get到异常,此时说明有任务创建失败,catch异常,并执行executerServce.shutDown(),接着将子任务线程中抛出的具体的异常信息返回给接入方。
此时几个任务中,未发生异常的任务将通过executerServce.isShutDown()判断是否已有任务产生异常,有则进行调用回滚方法(此处为防止网络问题,可通过mq异步调用回滚方法,基于mq的重试机制确保最终正确执行)。


7、课程创建子线程的实现描述:


创建失败,此时直接catch异常,并根据任务是否已创建判断是否执行回滚操作,接着抛出异常,此时主线程阻塞的take将获取到本FutureTask,并在执行get()方法后抛出异常,之后主线程如上面的“6、主线程实现描述#get到异常”中所说,主线程执行executerServce.shutDown(),“讲次创建任务” 和 “卷子创建任务” 将根据executerServce.isShutDown()判断为true后进行回滚。

创建成功,将通过countDownLatch.await()阻塞直到 “讲次创建任务” 和 “卷子创建任务” 都执行完countDownLatch.countDown()操作,countDownLatch.await()阻塞解除后将根据executerServce.isShutDown()判断主线程是否异常(即“讲次创建任务” 和 “卷子创建任务”是否有出现异常),有则调用回滚方法(此处为防止网络问题,可通过mq异步调用回滚方法,基于mq的重试机制确保最终正确执行),没有结束。


8、讲次创建子线程的实现描述:


创建失败,此时直接catch异常,并根据任务是否已创建判断是否执行回滚操作,执行countDownLatch.countDown()之后抛出异常,此时该异常将在主线程take到本FutureTask,并执行get()方法后抛出,之后主线程如“6、主线程实现描述#get到异常”中所说,主线程执行executerServce.shutDown(),“课程创建任务” 和 “卷子创建任务” 将根据executerServce.isShutDown()判断为true后进行回滚。

讲次创建成功后,将通过classTypeCreateAlready进行while轮询判断课程创建是否已经成功,classTypeCreateAlready为true,说明其余的两个任务都已经完成(只有讲次和卷子都成功并且调用countDownLatch后,课程创建才会从await()中继续执行并返回结果到主线程,设置classTypeCreateAlready为true),线程正常结束。
classTypeCreateAlready为false的话继续根据executerServce.isShutDown()判断主线程是否异常(即其余两个任务是否有出现异常),若executerServce.isShutDown()为true说明其他两个任务有异常,则讲次创建这里也将抛出异常,在catch中对创建成功的讲次信息进行回滚(此处为防止网络问题,可通过mq异步调用回滚方法,基于mq的重试机制确保最终正确执行)。

9、卷子创建子线程的实现描述:


创建失败,此时直接catch异常,并根据任务是否已创建判断是否执行回滚操作,接着执行countDownLatch.countDown()之后抛出异常,此时该异常将在主线程take到本FutureTask,并执行get()方法后抛出,之后主线程如“6、主线程实现描述#get到异常”中所说,主线程执行executerServce.shutDown(),“课程创建任务” 和 “讲次创建任务” 将根据executerServce.isShutDown()判断为true后进行回滚。

卷子创建成功后,将通过classTypeCreateAlready进行while轮询判断课程创建是否已经成功,classTypeCreateAlready为true,说明其余的两个任务都已经完成(只有讲次和卷子都成功并且调用countDownLatch后,课程创建才会从await()中继续执行并返回结果到主线程,设置classTypeCreateAlready为true),线程正常结束。
classTypeCreateAlready为false的话继续根据executerServce.isShutDown()判断主线程是否异常(即其余两个任务是否有出现异常),若executerServce.isShutDown()为true说明其他两个任务有异常,则卷子创建这里也将抛出异常,在catch中对创建成功的卷子信息进行回滚(此处为防止网络问题,可通过mq异步调用回滚方法,基于mq的重试机制确保最终正确执行)。

10、课程创建线程执行完毕后休眠1毫秒,是为了防止课程创建阻塞在countDownLatch.await()时,“讲次创建任务” 或 “卷子创建任务” 还在执行并接着产生异常,此时课程创建子任务执行executorService.isShutdown()一定要确保是在主线程take到FutureTask,并调用get()方法后,因此留出1毫秒主线程执行等待时间。


?


总结

以上即为课程创建本地分布式事务的实现,CompleteService是一个非常不错的并发类,当需要任务多线程执行,并且需要对多线程结果进行汇总时,将灰常有效。 而且还能实现只要其中一个线程异常,立即结束主流程并返回结果,你心动了么~


上述的实现中存在弊端,线程池未发挥真正的作用,之后进行了一轮优化版本。 之所以还写这个原始版本是因为当中用到了较多的并发类,能起到较好的学*效果。 如果哪位小伙伴知道如何优化,请在留言中告诉我哦,然后可以提出自己的思路。


优化版在另一篇blog?已给出 :?https://blog.csdn.net/y124675160/article/details/111185758??,以下的实现包含有CompleteService源码解析和使用以及CountDownLatch使用,感兴趣依然可以看一下。

相关推荐

最新更新

猜你喜欢