乐趣区

关于java:要我说多线程事务它必须就是个伪命题

这是 why 技术的第 74 篇原创文章

深夜怼文的我

别问,问就是不行

分布式事务你应该是晓得的。然而这个多线程事务 ……

没事,我缓缓给你说。

如图所示,有个小伙伴想要实现多线程事务。

这个需要其实我在不同的中央看到过很屡次,所以我才说:这个问题又呈现了。

那么有解决方案吗?

在此之前,我的答复都是十分的必定:毋庸置疑,必定是没有的。

为什么呢?

咱们先从实践下来推理一下。

来,首先我问你,事务的个性是什么?

这个不难吧?八股文必背内容之一,ACID 必须张口就来:

  • 原子性(Atomicity)
  • 一致性(Consistency)
  • 隔离性(Isolation)
  • 持久性(Durability)

那么问题又来了,你感觉如果有多线程事务,那么咱们毁坏了哪个个性?

多线程事务你也别想的多深奥,你就想,两个不同的用户各自发动了一个下单申请,这个申请对应的后盾实现逻辑中是有事务存在的。

这不就是多线程事务吗?

这种场景下你没有想过怎么别离去管制两个用户的事务操作吧?

因为这两个操作之间就是齐全隔离的,各自拿着各自的链接玩儿。

所以多个事务之间的最根本的准则是什么?

隔离性 。两个事务操作之间不应该互相烦扰。

而多线程事务想要实现的是 A 线程异样了。A,B 线程的事务一起回滚。

事务的个性外面就卡的死死的。所以,多线程事务从实践上就是行不通的。

通过理论指导实际,那么多线程事务的代码也就是写不进去的。

后面说到隔离性。那么请问,Spring 的源码外面,对于事务的隔离性是如何保障的呢?

答案就是 ThreadLocal。

在事务开启的时候,把以后的链接保留在了 ThreadLocal 外面,从而保障了多线程之间的隔离性:

能够看到,这个 resource 对象是一个 ThreadLocal 对象。

在上面这个办法中进行了赋值操作:

org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

其中的 bindResource 办法中,就是把以后链接绑定到以后线程中,其中的 resource 就是咱们刚刚说的 ThreadLocal:

就是每个线程外面都各自玩本人的,咱们不可能突破 ThreadLocal 的应用规定,让各个线程共享同一个 ThreadLocal 吧?

铁子,你要是这样去做的话,那岂不是走远了?

所以,无论从实践上,还是代码实现上,我都认为这个需要是不能实现的。

至多我之前是这样想的。

然而事件,稍稍的产生了一点点的变动。

说个场景,惯例实现

任何脱离场景探讨技术实现的行为都是耍流氓。

所以,咱们先看一下场景是什么。

假如咱们有一个大数据系统,每天指定工夫,咱们就须要从大数据系统中拉取 50w 条数据,对数据进行一个荡涤操作,而后把数据保留到咱们业务零碎的数据库中。

对于业务零碎而言,这 50w 条数据,必须全副落库,差一条都不行。要么就是一条都不插入。

在这个过程中,不会去调用其余的内部接口,也不会有其余的流程去操作这个表的数据。

既然说到一条不差了,那么对于大家直观而言,想到的必定是两个解决方案:

  1. for 循环中一条条的事务插入。
  2. 间接一条语句批量插入。

对于这种需要,开启事务,而后在 for 循环中一条条的插入能够说是十分 low 的解决方案了。

效率十分的低下,给大家演示一下。

比方,咱们有一个 Student 表,表构造非常简单,如下:

`CREATE TABLE student` (
  id bigint(63) NOT NULL AUTO_INCREMENT,
  name varchar(32) DEFAULT NULL,
  home varchar(64) DEFAULT NULL,
  PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

在咱们的我的项目中,咱们通过 for 循环插入数据,同时该办法上有 @Transactional 注解:

num 参数是咱们通过前端申请传递过去的数据,代表要插入 num 条数据:

这种状况下,咱们能够通过上面的链接,模仿插入指定数量的数据:

http://127.0.0.1:8081/insertOneByOne?num=xxx

我尝试了把 num 设置为 50w,让它缓缓的跑着,然而我还是太年老了,等了十分长的工夫都没有等到后果。

于是我把 num 改为了 5000,运行后果如下:

insertOneByOne 执行耗时:133449ms,num=5000

一条条的插入 5000 条数据,耗时 133.5 s 的样子。

依照这个速度,插入 50w 条数据得 13350s,大略也是这么多小时:

这谁顶得住啊。

所以,这计划领有微小的优化空间。

比方咱们优化为这样批量插入:

其对应的 sql 语句是这样的:

insert into table ([列名],[列名]) VALUES ([列值],[列值]), ([列值],[列值]);

咱们还是通过前端接口调用:

当咱们的 num 设置为 5000 的时候,我页面刷新了 10 次,你看耗时基本上在 200ms 毫秒以内:

从 133.5s 到 200ms,敌人们,这是什么货色?

这是质的飞跃啊。性能晋升了近 667 倍的样子。

为什么批量插入能有这么大的飞跃呢?

你想啊,之前 for 循环插入,尽管 SpringBoot 2.0 默认应用了 HikariPool,连接池外面默认给你搞 10 个连贯。

然而你只须要一个连贯,开启一次事务。这个不耗时。

耗时的中央是你 5000 次 IO 呀。

所以,耗时长是必然的。

而批量插入只是一条 sql 语句,所以只须要一个连贯,还不须要开启事务。

为啥不必开启事务?

你一条 sql 开启事务有锤子用啊?

那么,如果咱们一口气插入 50w 条数据,会是怎么样的呢?

来,搞一波,试一下:

http://127.0.0.1:8081/insertBatch?num=500000

能够看到抛出了一个异样。而且错误信息十分的清晰:

`Packet for query is too large (42777840 > 1048576). You can change this value on the server by setting the max_allowed_packet’ variable.; nested exception is com.mysql.jdbc.PacketTooBigException: Packet for query is too large (42777840 > 1048576).You can change this value on the server by setting the max_allowed_packet’ variable.
`

说你这个包太大了。能够通过设置 max_allowed_packet 来扭转包大小。

咱们能够通过上面的语句查问以后的配置大小:

select @@max_allowed_packet;

能够看到是 1048576,即 1024*1024,1M 大小。

而咱们须要传输的包大小是 42777840 字节,大略是 41M 的样子。

所以咱们须要批改配置大小。

这个中央也给大家提了个醒: 如果你的 sql 语句十分大,外面有大字段,记得调整一下 mysql 的这个参数。

能够通过批改配置文件或者间接执行 sql 语句的形式进行批改。

我这里就应用 sql 语句批改为 64M:

set global max_allowed_packet = 1024*1024*64;

而后再次执行,能够看到插入胜利了:

50w 的数据,74s 的样子。

数据要么全副提交,要么一条也没有,需要也实现了。

工夫上呢,是有点长,然而如同也想不到什么好的晋升计划。

那么咱们怎么还能再缩短点工夫呢?

骚想法呈现了

我能想到的,只能是祭出多线程了。

50w 数据。咱们开五个线程,一个线程解决 10w 数据,没有异样就保留入库,呈现问题就回滚。

这个需要很好实现。分分钟就能写进去。

然而再加上一个需要: 这 5 个线程的数据,如果有一个线程呈现问题了,须要全副回滚。

顺着思路缓缓撸,咱们发现这个时候就是所谓的多线程事务了。

我之前说齐全不可能实现是因为提到事务我就想到了 @Transactional 注解去实现了。

咱们只须要正确应用它,而后关系业务逻辑即可,不须要也基本插手不了事务的开启和提交或者回滚。

这种代码的写法咱们叫做申明式事务。

和申明式事务对应的就是编程式事务了。

通过编程式事务,咱们就能齐全掌控事务的开启和提交或者回滚操作。

能想到编程式事务,这事基本上就成了一半了。

你想,首先咱们有一个全局变量为 Boolean 类型,默认为能够提交。

在子线程外面,咱们能够先通过编程式事务开启事务,而后插入 10w 条数据后,然而不提交。同时通知主线程,我这边筹备好了,进入期待。

如果子线程外面呈现了异样,那么我就通知主线程,我这边出问题了,而后本人进行回滚。

最初主线程收集到了 5 个子线程的状态。

如果有一个线程呈现了问题,那么设置全局变量为不可提交。

而后唤醒所有期待的子线程,进行回滚。

依据下面的流程,写出模仿代码就是这样的,大家能够间接复制进去运行:

`public class MainTest {
    // 是否能够提交
    public static volatile boolean IS_OK = true;
    public static void main(String[] args) {
        // 子线程期待主线程告诉
        CountDownLatch mainMonitor = new CountDownLatch(1);
        int threadCount = 5;
        CountDownLatch childMonitor = new CountDownLatch(threadCount);
        // 子线程运行后果
        List<Boolean> childResponse = new ArrayList<Boolean>();
        ExecutorService executor = Executors.newCachedThreadPool();
        for (int i = 0; i < threadCount; i++) {
            int finalI = i;
            executor.execute(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + “:开始执行 ”);
// if (finalI == 4) {
// throw new Exception(“ 出现异常 ”);
// }
                    TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(1000));
                    childResponse.add(Boolean.TRUE);
                    childMonitor.countDown();
                    System.out.println(Thread.currentThread().getName() + “:准备就绪, 期待其余线程后果, 判断是否事务提交 ”);
                    mainMonitor.await();
                    if (IS_OK) {
                        System.out.println(Thread.currentThread().getName() + “:事务提交 ”);
                    } else {
                        System.out.println(Thread.currentThread().getName() + “:事务回滚 ”);
                    }
                } catch (Exception e) {
                    childResponse.add(Boolean.FALSE);
                    childMonitor.countDown();
                    System.out.println(Thread.currentThread().getName() + “:出现异常, 开始事务回滚 ”);
                }
            });
        }
        // 主线程期待所有子线程执行 response
        try {
            childMonitor.await();
            for (Boolean resp : childResponse) {
                if (!resp) {
                    // 如果有一个子线程执行失败了,则扭转 mainResult,让所有子线程回滚
                    System.out.println(Thread.currentThread().getName()+”: 有线程执行失败,标记位设置为 false”);
                    IS_OK = false;
                    break;
                }
            }
            // 主线程获取后果胜利,让子线程开始依据主线程的后果执行(提交或回滚)
            mainMonitor.countDown();
            // 为了让主线程阻塞,让子线程执行。
            Thread.currentThread().join();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
`

在所有子线程都失常的状况下,输入后果是这样的:

从后果看,是合乎咱们的预期的。

假如有子线程呈现了异样,那么运行后果是这样的:

一个线程出现异常,全副线程都进行回滚,这样看来也是合乎预期的。

如果你依据后面的需要写出了这样的代码,那么祝贺你,一不留神实现了一个相似于两阶段提交(2PC)的一致性协定。

我后面说的能想到编程式事务,这事基本上就成了一半了。

而另外一半,就是两阶段提交(2PC)。

依瓢画葫芦

有了后面的瓢,你照着画个葫芦不是很简略的事件吗?

就不大段上代码了, 示例代码能够点击这里获取到 ,所以我这里截个图吧:

下面的代码应该是十分好了解的,开启五个线程,每个线程插入 10w 条数据。

这个不用说,用脚趾头想也能晓得,必定是比一次性批量插入 50w 条数据快的。

至于快多少,不废话了,间接看执行成果吧。

因为咱们的 controller 是这样的:

所以调用链接:

http://127.0.0.1:8081/batchHandle

输入后果如下:

还记得咱们批量插入的耗时吗?

73791ms。

从 73791ms 到 15719ms。快了 58s 的样子。

曾经十分不错了。

那么如果是某个线程抛出了异样呢?比方这样:

咱们看看日志输入:

通过日志剖析,看起来也是符合要求的。

而从读者反馈的理论测试成果来看,也是十分显著的:

真的符合要求吗?

符合要求,只是看起来而已。

教训老道的读者敌人们必定早就看到问题所在了。曾经把手举得高高的:老师,这题我晓得。

我之前说了,这个实现形式实际上就是编程式事务配合二阶段提交(2PC)应用。

漏洞就出在 2PC 上。

就像我和读者探讨这样的:

不能再往后扯了,再往后就是 3PC,TCC,Seata 这一套分布式事务的货色了。

这套货色写下来,就得上万字了。所以我从海神那边转了一篇文章,放在第二条推送外面了。如果大家有趣味的能够去看一下。干货满满。

其实当咱们把一个个子线程了解为微服务中的一个个子系统的时候,这就是一个分布式事务的场景了。

而咱们拿进去的解决方案,并不是一个完满的解决方案。

尽管,从某种角度上,咱们绕开了事务的隔离性,然而有肯定概率呈现数据一致性问题,尽管概率比拟小。

所以我称之为这种计划叫做:基于运气编程,用运气换工夫。

注意事项

对于下面的代码,其实还有几个须要留神的中央。

给大家提个醒。

第一个 :启用多少线程进行调配数据插入,这个参数是能够进行调整的。

比方我批改为 10 个线程,每个线程插入 5w 条数据。那么执行工夫又快了 2s:

然而肯定记得不是越大越好,同时记得调整数据库连接池的最大连接数。不然白搭。

第二个 :正是因为启动多少线程是能够进行调整的,甚至是能够每次进行计算的。

那么必须要留神的一个问题是不能让任何一个工作进入队列外面。一旦进入队列,程序立马就凉。

你想,如果咱们须要开启 5 个子线程,然而外围线程数只有 4 个,有一个工作进入队列了。

那么这 4 个外围线程会始终阻塞住,期待主线程唤醒。

而主线程这个时候在干什么?

在等 5 个线程的运行后果,然而它只能收集到 4 个后果。

所以它会始终等上来。

第三个 :这里是多个线程开启了事务在往表里插入数据,谨防数据库死锁。

第四个 :留神程序外面的代码,countDown 装置规范写法上是要放到 finally 代码块外面的,我这里为了截图的好看度,省去了这个步骤:

你如果真的要用,得留神一下。而且这个 finally 你得想分明了写,不是轻易写的。

第五个 :我这里只是提供一个思路,而且它也基本不是什么多线程事务。

也再次证实了,多线程事务就是一个伪命题。

所以我给出一个基于运气的伪一致性的答复也不过分吧。

第六个 :多线程事务换个角度想,能够了解为分布式事务。,能够借助这个案例去理解分布式事务。然而解决分布式事务的最好的办法就是:不要有分布式事务!

而解决分布式事务的绝大部分落地计划都是:最终一致性。

性价比高,大多数业务上也能承受。

第七个 :这个解决方案你要拿到生产用的话,记得先和业务共事沟通好,能不能承受这种状况。速度和平安之间的两难抉择。

同时本人留好人工修数的接口:

最初说一句

满腹经纶,难免会有纰漏,如果你发现了谬误的中央,能够在留言区提出来,我对其加以批改。感谢您的浏览,我保持原创,非常欢送并感谢您的关注。

我是 why,一个被代码耽搁的文学创作者,不是大佬,然而喜爱分享,是一个又暖又有料的四川好男人。

还有,欢送关注我呀。

退出移动版