流计算中的死锁

作者：Xiao8

2022 年 6 月 18 日
本文字数：3342 字
阅读完需：约 11 分钟

作者：Damon

博客：http://www.damon8.cn

交个朋友之猿天地 | 微服务 | 容器化 | 自动化

为什么流计算过程不能有环

我们从一个简单的流计算过程开始，这个流计算过程的 DAG 如下图:

DAG 描述了一个最简单的流计算过程，步骤 A 的输出给步骤 B 进行处理。这个流计算过程用 CompletableFuture 实现非常简单。如下所示：

ExecutorService AExecutor = new BackPressureExecutor(
        "AExecutor", 1, 1, 1, 10, 1);
ExecutorService BExecutor = new BackPressureExecutor(
        "BExecutor", 1, 1, 1, 10, 1);
AtomicLong itemCounter = new AtomicLong(0L);
String stepA() {
    String item = String.format("item%d", itemCounter.getAndDecrement());
    logger.info(String.format("stepA item[%s]", item));
    return item;
}
void stepB(String item) {
    logger.info(String.format("stepB item[%s]", item));
    sleep(10); // 睡眠一会，故意让 stepB 处理得比 stepA 慢
}
void demo1() {
    while (!Thread.currentThread().isInterrupted()) {
        CompletableFuture
                .supplyAsync(this::stepA, this.AExecutor)
                .thenAcceptAsync(this::stepB, this.BExecutor);
    }
}

复制代码

上面的代码中，步骤 A 和步骤 B 使用了两个不同的执行器，即 AExecutor 和 BExecutor 。并且为了避免 OOM 问题，我们使用的执行器都是带反向压力功能的 BackPressureExecutor。

上面的程序运行起来没有任何问题。即便我们明确通过 sleep 函数，让 stepB 的处理速度只有 stepA 的十分之一，上面的程序都能够长时间的稳定运行（stepA 和 stepB 会不断打印出各自的处理结果，并且绝不会出现 OOM 问题）。

但是现在，我们需要对图 1 的 DAG 稍微做点变化，让 B 在处理完后，将其结果重新输入给自己再处理一次。这种处理逻辑，在实际开发中也会经常遇到，比如 B 在处理失败时，就将处理失败的任务，重新添加到自己的输入队列，从而实现失败重试的功能。修改后的 DAG 如下图所示:

很明显，上面的 DAG 在步骤 B 上形成了一个“环”，因为有一条从 B 开始的有向线段，重新指向了 B 自己。相应的，前面的代码也需要稍微做点调整，改成下面的方式：

void demo1() {
    while (!Thread.currentThread().isInterrupted()) {
        CompletableFuture
                .supplyAsync(this::stepA, this.AExecutor)
                .thenApplyAsync(this::stepB, this.BExecutor)
                .thenApplyAsync(this::stepB, this.BExecutor);
    }
}

复制代码

上面的代码中，我们增加了一次 thenApplyAsync 调用，用于将 stepB 的输出重新作为其输入。需要注意的是，由于第二次 stepB 调用后没有再设置后续步骤，所以，虽然 DAG 上“有环”，但 stepB 并不会形成死循环。

上面这段代码，初看起来并没什么问题，毕竟就是简单地新增了一个“重试”的效果嘛。但是，如果你实际运行上面这段代码就会发现，只需要运行不到 1 秒钟，上面这段程序就会“卡”住，之后控制台会一动不动，没有一条日志打印出来。

这是怎么回事呢？事实上这就是因为程序已经“死锁”了！

流计算过程死锁分析

说到“死锁”，你一定会想到“锁”的使用。一般情况下之所以会出现“死锁”，主要是因为我们使用锁的方式不对，比如使用了不可重入锁，或者使用多个锁时出现了交叉申请锁的情况。这种情况下出现的“死锁”问题，我们确确实实看到了“锁”的存在。

但当我们在使用流计算编程时，你会发现，“流”的编程方式已经非常自然地避免了“锁”的使用，也就是说我们并不会在“流”处理的过程中用到“锁”。这是因为，当使用“流”时，被处理的对象依次从上游流到下游。当对象在流到某个步骤时，它是被这个步骤的线程唯一持有，因此不存在对象竞争的问题。

但这是不是就说流计算过程中不会出现“死锁”问题呢？不是的。最直接的例子就是前面的代码，我们根本就没有用到“锁”，但它还是出现了“死锁”的问题。

所以，为什么会出现“死锁”呢？这里就需要我们仔细分析下了。下面的图 3 描绘了图 2 中的流计算过程之所以会发生死锁的原因。

在图中，整个流计算过程有 A 和 B 这两个步骤，并且具备“反向压力”能力。这时候，如果 A 的输出已经将 B 的输入队列占满，而 B 的输出又需要重新流向 B 的输入队列，那么由于“反向压力”的存在，B 会一直等到其输入队列有空间可用。而 B 的输入队列又因为 B 在等待，永远也不会有空间被释放，所以 B 会一直等待下去。同时，A 也会因为 B 的输入队列已满，由于反向压力的存在，它也只能不停地等待下去。

如此一来，整个流计算过程就形成了一个死锁，A 和 B 两个步骤都会永远等待下去，这样就出现了我们前边看到的程序“卡”住现象。

形成“环”的原因

在图 2 所示的 DAG 中，我们是因为需要让 stepB 失败重试，所以“随手”就让 stepB 将其输出重新作为输入重新执行一次。这姑且算是一种比较特殊的需求吧。

但在实际开发过程中，我们的业务逻辑明显是可以分为多个依次执行的步骤，用 DAG 画出来时，也是“无环”的。但在写代码时，有时候一不小心，也会无意识地将一个本来无环的 DAG，实现成了有环的过程。

比如：业务逻辑本来是 A 到 B 到 C 这样的“无环”图，结果由于我们给这三个不同的步骤，分配了同一个执行器 executor，实际实现的流计算过程就成了一个“有环”的过程。

在这个“有环”的实现中，只要任意一个步骤的处理速度比其他步骤慢，就会造成执行器的输入队列占满。一旦输入队列占满，由于反向压力的存在，各个步骤的输出就不能再输入到队列中。最终，所有执行步骤将会阻塞，也就形成了死锁，整个系统也被“卡”死。

如何避免死锁

所以，我们在流计算过程中，应该怎样避免死锁呢？其实很简单，有三种方法。

一是不使用反向压力功能。只需要我们不使用反向压力功能，即使业务形成“环”了，也不会死锁，因为每个步骤只需要将其输出放到输入队列中，不会发生阻塞等待，所以就不会死锁了。但很显然，这种方法禁止使用。毕竟，没有反向压力功能，就又回到 OOM 问题了，这是万万不可的！

二是避免业务流程形成“环”。这个方法最主要的作用，是指导我们在设计业务流程时，不要将业务流程设计成“有环”的了。否则如果系统有反向压力功能的话，容易出现类似于图 3 的死锁问题。

三是千万不要让多个步骤使用相同的队列或执行器。这个是最容易忽略的问题，特别是一些对异步编程和流计算理解不深的开发人员，最容易给不同的业务步骤分配相同的队列或执行器，在不知不觉中就埋下了死锁的隐患。

总的来说，在流计算过程中，反向压力功能是必不可少的，为了避免“死锁”的问题，流计算过程中的任何一个步骤，它的输出绝不能再重新流回作为它的输入。

只需要注意以上几点，你就可以放心大胆地使用“流”式编程了，而且不用考虑“锁”的问题。由于没有了竞态问题，这既可以简化你编程的过程，也可以给程序带来显著的性能提升。

精彩推荐

结束福利

开源实战利用 k8s 作微服务的架构设计代码：

欢迎大家 star，多多指教。

关于作者

笔名：Damon，技术爱好者，长期从事 Java 开发、Spring Cloud 的微服务架构设计，以及结合 Docker、K8s 做微服务容器化，自动化部署等一站式项目部署、落地。目前主要从事基于 K8s 云原生架构研发的工作。Golang 语言开发，长期研究边缘计算框架 KubeEdge、调度框架 Volcano 等。公众号 程序猿Damon 发起人。个人微信 MrNull008，个人网站：Damon | Micro-Service | Containerization | DevOps，欢迎來撩。

欢迎关注：InfoQ

欢迎关注：腾讯自媒体专栏