Ohhnews

分类导航

$ cd ..
DZone Java原文

Java虚拟线程在高负载下的三大生产限制

#java#虚拟线程#并发控制#限流#资源管理

虚拟线程让我们可以保留普通的阻塞代码，同时让更多请求并发等待。然而，在负载下，这会引出三个关键问题：

  • 我们允许多少工作进入？ 线程再便宜，也可能堆积出代价高昂的队列。
  • 什么应该放在锁内？ 持有锁的线程在睡眠时，仍会让其他所有线程等待。
  • 谁来停止被遗弃的工作？ 调用方超时并不会取消其任务。

下面的示例用各自独立的前后对比程序来分别说明这些问题。每个程序都是完整的：把其代码块保存为所示文件名，并在不带参数的情况下运行随附命令。你只需要 JDK 25。它们使用 JDK 25 的紧凑源文件语法，因此无需类声明、构建工具、外部库或预览标志。

首先：更多重叠的等待

设想 1,000 个任务，每个任务等待 100 毫秒，且没有依赖项限制它们的并发：

执行模型同时等待数理想耗时
50 个平台线程组成的线程池502,000 毫秒
每个任务一个虚拟线程，所有等待重叠1,000100 毫秒
虚拟线程限制为 50 个并发等待502,000 毫秒

这些是算术估算，忽略了启动和调度开销。使用 50 个工作线程时，这些等待需要二十批。让全部 1,000 个等待重叠，则只需一批。将虚拟线程并发限制为 50，二十批的结果又会回来。虚拟线程让更多相互独立的等待得以重叠。它们并不会更快地执行 CPU 指令。虚拟线程运行在一个称为载体的平台线程上。许多阻塞式 JDK 操作（包括套接字 I/O 和 sleep）会让虚拟线程在等待时卸载。它的栈和被引用的对象仍保留在堆上，而载体可以运行其他内容。

[LOADING...]

（M:N 调度模型：许多虚拟线程映射到较小的操作系统载体线程池）

调度器的默认目标并行度是 JVM 可用的处理器数量。某些阻塞操作仍会占用载体；即使在 JDK 25 上，本地调用和外部函数调用也可能固定虚拟线程。等待会使用更少的操作系统线程，但仍然消耗内存和资源。

1. 限制准入：八个槽位无法同时服务所有人

设想一个依赖项有八个连接。每次调用耗时 80 毫秒。其理想容量为 8 / 0.08 = 每秒 100 次调用。虚拟线程可以廉价地等待连接，但它们无法再创建一个连接。发送 100 个请求的突发流量。首先，全部接受。然后，用 32 的准入上限重复一次：最多八个正在使用依赖项，另有 24 个在等待它。

首先，接受每个请求。保存为 UnboundedAdmission.java：

$ java
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.concurrent.*;

void main() throws Exception {
    var connections = new Semaphore(8);
    var results = new ArrayList<Future<Boolean>>();
    Instant now = Instant.now();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < 100; i++) {
            results.add(executor.submit(() -> {
                connections.acquire();
                try {
                    Thread.sleep(Duration.ofMillis(80)); // the dependency call
                } finally {
                    connections.release();
                }
                return null;
            }));
        }
    }
    // wait for all work to finish
    for (var result : results) result.get();

    IO.println("completed=%d rejected=0 elapsed_ms=%d".formatted(
            results.size(),
            Duration.between(now, Instant.now()).toMillis()));
}

运行它：java UnboundedAdmission.java

每个请求都会等待，直到有连接可用。现在添加一个准入信号量，让超额请求立即被拒绝。保存为 BoundedAdmission.java：

$ java
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.concurrent.*;

void main() throws Exception {
    var connections = new Semaphore(8);
    var admission = new Semaphore(32);
    var results = new ArrayList<Future<Boolean>>();
    Instant now = Instant.now();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < 100; i++) {
            results.add(executor.submit(() -> {
                if (!admission.tryAcquire()) return false;
                try {
                    connections.acquire();
                    try {
                        Thread.sleep(Duration.ofMillis(80)); // the dependency call
                    } finally {
                        connections.release();
                    }
                    return true;
                } finally {
                    admission.release();
                }
            }));
        }
    }
    // wait for accepted work to finish
    int completed = 0;
    for (var result : results) if (result.get()) completed++;

    IO.println("completed=%d rejected=%d elapsed_ms=%d".formatted(
            completed, 100 - completed,
            Duration.between(now, Instant.now()).toMillis()));
}

运行它：java BoundedAdmission.java

在 Oracle GraalVM 25.0.2 上的一次运行产生了：

准入完成数拒绝数耗时
不限制10001,062 毫秒
上限为 323268330 毫秒

这里有两个不同的限制。connections.acquire() 等待八个服务槽位之一。admission.tryAcquire() 会立即接纳请求或返回 false，并记录一次拒绝。它的许可会在请求等待连接并执行调用期间一直保持占用。没有准入控制时，全部 100 个请求在十三批内完成，耗时约一秒。有了这道闸门后，一次快速突发通常接纳 32 个并拒绝 68 个；四批大约耗时 320 毫秒。有界运行结束得更早，因为它拒绝了超额工作，避免了内存耗尽。即使每个线程缓存只有 10 KB，乘以 100,000 个虚拟线程，仍然会占用 1 GB 堆空间。要点：在把工作加入你需要限制的队列之前，先拒绝它。

2. 缩短锁：保护更新

现在使用四十个任务。每个任务等待 80 毫秒，并递增一个共享的完成计数器。这里没有连接限制：这些等待相互独立，因此锁是唯一的瓶颈。首先，把等待和计数器更新都放在锁内。保存为 WideLock.java：

$ java
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.concurrent.*;

final Object lock = new Object();
int completed;

void main() throws Exception {
    Instant now = Instant.now();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var tasks = new ArrayList<Future<Boolean>>();
        for (int i = 0; i < 40; i++) {
            tasks.add(executor.submit(() -> {
                synchronized (lock) {
                    Thread.sleep(Duration.ofMillis(80));
                    completed++;
                }
                return null;
            }));
        }
        for (var task : tasks) task.get();
    }
    if (completed != 40) throw new AssertionError("Lost an update");
    IO.println("completed=%d elapsed_ms=%d".formatted(
            completed,
            Duration.between(now, Instant.now()).toMillis()));
}

运行它：java WideLock.java

现在把等待移到锁外，同时保持计数器更新受保护。保存为 NarrowLock.java：

$ java
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.concurrent.*;

final Object lock = new Object();
int completed;

void main() throws Exception {
    Instant now = Instant.now();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var tasks = new ArrayList<Future<Boolean>>();
        for (int i = 0; i < 40; i++) {
            tasks.add(executor.submit(() -> {
                Thread.sleep(Duration.ofMillis(80));
                synchronized (lock) {
                    completed++;
                }
                return null;
            }));
        }
        for (var task : tasks) task.get();
    }
    if (completed != 40) throw new AssertionError("Lost an update");
    IO.println("completed=%d elapsed_ms=%d".formatted(
            completed,
            Duration.between(now, Instant.now()).toMillis()));
}

运行它：java NarrowLock.java

在同一个 JDK 上，两次运行产生了：

锁范围完成数耗时
等待和更新都在锁内403,255 毫秒
只有更新在锁内4087 毫秒

当等待位于锁内时，四十个 80 毫秒的等待至少需要 3,200 毫秒。让其他三十九个线程更廉价地挂起，并不能让它们越过持有锁的线程。当等待位于锁外时，这些等待就会重叠。在 JDK 25 上，synchronized 完全适合这个演示，因为 JDK 24 移除了与监视器相关的虚拟线程固定。这里的问题不是平台线程固定；而是基本的线程串行化。规则是：找出需要保护的状态转换，然后把相互独立的慢操作留在它之外。

3. 取消工作：超时只结束等待

使用一个耗时 1,800 毫秒的任务。其调用方只愿意等待 200 毫秒。首先，让调用方超时但不取消任务。保存为 TimeoutOnly.java：

$ java
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.*;

void main() throws Exception {
    Instant now = Instant.now();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var task = executor.submit(() -> {
            try {
                Thread.sleep(Duration.ofMillis(1_800));
                IO.println("Work completed");
                return "done";
            } finally {
                IO.println("Cleanup ran");
            }
        });
        try {
            task.get(200, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            IO.println("Caller timed out");
        }
    } // close() still waits for the task to exit

    IO.println("scope_exited_ms=%d".formatted(
            Duration.between(now, Instant.now()).toMillis()));
}

运行它：java TimeoutOnly.java

现在在超时处理程序中添加 task.cancel(true)。保存为 TimeoutWithCancellation.java：

$ java
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.*;

void main() throws Exception {
    Instant now = Instant.now();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var task = executor.submit(() -> {
            try {
                Thread.sleep(Duration.ofMillis(1_800));
                IO.println("Work completed");
                return "done";
            } finally {
                IO.println("Cleanup ran");
            }
        });
        try {
            task.get(200, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            IO.println("Caller timed out");
            task.cancel(true); // <--- Explicit cancellation
        }
    } // close() waits for the task to exit, even after cancellation

    IO.println("scope_exited_ms=%d".formatted(
            Duration.between(now, Instant.now()).toMillis()));
}

运行它：java TimeoutWithCancellation.java

观察到的结果如下：

示例是否打印 Work completed？是否打印 Cleanup ran？作用域退出时间
仅超时是是1,805 毫秒
超时加取消否是207 毫秒

没有取消时，get() 在 200 毫秒后抛出异常，但任务仍继续睡眠。它最终会在 1.6 秒后完成工作。如果这个被遗弃的工作持有数据库连接或准入许可，那么该资源将在整个睡眠期间泄漏。取消是协作式的。检测超时、请求取消，并等待清理完成。这三个部分都很重要。

预告：结构化并发解决方案

在上面的最后一个示例中，你会注意到父线程被迫对任务的生命周期进行微观管理。它等待、捕获 TimeoutException、显式调用 cancel(true)，然后祈祷清理逻辑会运行。但如果我们可以反转这种控制呢？借助 JDK 的预览版 StructuredTaskScope API，我们可以把线程的生命周期绑定到词法作用域本身，让取消成为一种自动的架构保证，而不是手动样板代码：我将在下一篇文章中探讨 StructuredTaskScope 如何彻底消除资源泄漏和被遗弃的工作。敬请期待！

总结

迁移到虚拟线程时，请把这三个问题带回你的服务中：

  1. 等待在哪里堆积？ 在那个队列增长之前限制准入。
  2. 是什么让工作串行化？ 把相互独立的等待留在保护状态的锁之外。
  3. 谁负责未完成的工作？ 把截止时间与协作式取消和清理连接起来。

这里的程序使用 sleep 让每种效果都可见，但当你的等待对象是数据库、套接字或合作伙伴 API 时，完全相同的问题依然适用。