论虚拟线程与 Kotlin IO 协程：资源开销、时长、高并发表现及适用场景与技术选型思考

在现代并发编程中，虚拟线程（由 Java 20+ 引入）和 Kotlin IO 协程（基于 Dispatchers.IO）是两种高效处理异步任务的技术框架。在资源开销、时长（特别是长时间 IO）、高并发场景表现、以及何时选择合适方案等方面各有特点。本文将逐一展开对比分析。

1. 资源开销对比

虚拟线程和协程的核心开销差异源于其底层设计原理：

[注1] 1MB 是虚拟地址空间最大预留值（非物理内存提交量），实际物理内存按使用量提交，浅调用栈的虚拟线程实际仅占几十~几百KB。实测 10,000 个虚拟线程总增量约 3 MB，均摊约 0.3 KB/线程；而 10,000 个协程的增量几乎不可测量（< 0.1 MB，均摊约 0.01 KB/协程）。协程约 30 倍内存优势，但虚拟线程的绝对内存开销其实非常可控。

[注2] 此为微观层面的挂起/恢复指令开销。宏观层面的完整调度往返（含 JVM、线程池排队、GC 等）在微秒级，详见[注3]。

[注3] 实测数据来自 JDK 21.0.5, Kotlin 2.3.21, 12核CPU 环境（BenchmarkVerification.kt），完整数据见文末附录 A。

资源模型公式：设任务数 NNN
协程内存开销=N⋅Sdynamic-stack(Sdynamic-stack∈[0.002,0.5]MB) \text{协程内存开销} = N \cdot S_{\text{dynamic-stack}} \quad (S_{\text{dynamic-stack}} \in [0.002, 0.5] \text{MB}) 协程内存开销=N⋅Sdynamic-stack​(Sdynamic-stack​∈[0.002,0.5]MB)
虚拟线程内存开销=N⋅Sfixed-stack(Sfixed-stack=1MB) \text{虚拟线程内存开销} = N \cdot S_{\text{fixed-stack}} \quad (S_{\text{fixed-stack}} = 1\text{MB}) 虚拟线程内存开销=N⋅Sfixed-stack​(Sfixed-stack​=1MB)
其中当 N>104N > 10^4N>104 时，虚拟线程内存快速膨胀。需要注意该公式中的 1MB 为虚拟地址预留的上界，物理内存增量远小于此值。

2. 时长分析：长时间 IO 的定义与影响

在并发模型中，长时间 IO 指一次阻塞操作耗时 超过 10ms10\text{ms}10ms（如网络调用 T>20msT > 20\text{ms}T>20ms 、磁盘读写 T>50msT > 50\text{ms}T>50ms ），当 Tio>TctxT_{\text{io}} > T_{\text{ctx}}Tio​>Tctx​（系统上下文切换时间约 1–10μs1–10\mu\text{s}1–10μs)时，系统效率由TioT_{\text{io}}Tio​主导，影响包括：

• 协程响应模型：
  延迟=Tio线程池大小+Cresume 延迟 = \frac{T_{\text{io}}}{线程池大小} + C_{\text{resume}}延迟=线程池大小Tio​​+Cresume​
• 虚拟线程机制：
  延迟=OS调度等待时间+系统调度（Tschedule） 延迟 = \text{OS调度等待时间} + {\small系统调度}（T_{\text{schedule}}）延迟=OS调度等待时间+系统调度（Tschedule​）
  当Tio>10msT_{\text{io}} > 10\text{ms}Tio​>10ms时，虚拟线程不阻塞宿主线程资源能成优势[注1]。

实测补充：实测数据支持 Tio>10msT_{\text{io}} > 10\text{ms}Tio​>10ms 作为临界区间。当 IO=5ms 时，Dispatchers.IO（默认）吞吐量仍领先虚拟线程；当 IO=50ms 时，虚拟线程开始在高并发下（N≥5,000）反超约 10~35%；当 IO=200ms 时，虚拟线程领先约 3~10%。当 IO 达到 1000ms 以上时，所有调度器的吞吐量趋同（IO 耗时完全主导）。详见附录 A。

3. 高并发场景下的表现对比

在高压力并发下（任务数 N≥104N \geq 10^4N≥104），系统表现受阻塞时长和调度效率影响：

• 虚拟线程：吞吐量模型：
  虚拟线程吞吐量=NTctx+Tio \text{虚拟线程吞吐量} = \frac{N}{T_{\text{ctx}} + T_{\text{io}}}虚拟线程吞吐量=Tctx​+Tio​N​
  如 Tio=100msT_{\text{io}}=100\text{ms}Tio​=100ms 时可达 9.8K9.8K9.8K ops/s[注1]，利用 OS 自动释放资源特性提升并发。
• Kotlin IO 协程：吞吐量模型：
  协程吞吐量=N⋅KactiveTexec+Tio \text{协程吞吐量} = \frac{N \cdot K_{\text{active}}}{T_{\text{exec}} + T_{\text{io}}}协程吞吐量=Texec​+Tio​N⋅Kactive​​
  其中 KactiveK_{\text{active}}Kactive​ 是调度器资源上限，结构化并发模型避免资源滥用，在高竞争下略低（如7.2K7.2K7.2K ops/s）。

实测补充：上述吞吐量公式的结构正确，但 KactiveK_{\text{active}}Kactive​ 的具体值受 Dispatchers.IO 默认并行度限制 max(64, CPU核数) 影响。实测 N=10,000, T_io=100ms 时，IO(默认) 约 62.8K ops/s，原生 VT 约 83.8K ops/s（领先 33%）。当解除 IO 并行度限制后（limitedParallelism(MAX)），IO(无限制) 约 82.8K ops/s，与 VT 接近。这说明 KactiveK_{\text{active}}Kactive​ 的核心限制来自线程池并行度而非协程调度本身。

4. 何时选择虚拟线程或 Kotlin IO 协程

基于临界点 Tio-critT_{\text{io-crit}}Tio-crit​ 模型：
Tio-crit=Tctx⋅Cthread-costKactive⋅Rmem-cost T_{\text{io-crit}} = \frac{T_{ctx} \cdot C_{\text{thread-cost}}}{K_{\text{active}} \cdot R_{\text{mem-cost}}}Tio-crit​=Kactive​⋅Rmem-cost​Tctx​⋅Cthread-cost​​
（Rmem-costR_{\text{mem-cost}}Rmem-cost​ 是内存成本比，典型值 R≈12.5R \approx 12.5R≈12.5）。
具体建议：

• 虚拟线程适用场景：
  • 长阻塞操作占主导（如 Tio>1.5msT_{\text{io}} >1.5\text{ms}Tio​>1.5ms)
  • GC 开销可接受，但需高响应吞吐，如外部服务调用密集场景
  • 目标处理 N>104N >10^4N>104 任务
• 协程适用场景：
  • IO 时间较短或混合负载系统
  • 内存资源敏感，如嵌入式或 Android 移动端
  • 需避免系统突破任务池上限，实现优雅降级
  • 混合 CPU+IO 负载：当 CPU 计算占比 >20% 时，Dispatchers.IO 的平台线程池 work-stealing 机制使其在混合负载下显著优于虚拟线程（实测领先 15%~146%）[注4]

[注4] 混合负载是生产中最常见的真实场景（如：接收请求 → CPU计算/校验 → 调用外部服务IO → CPU组装响应）。实测 5 组混合场景（CPU 从 2ms 到 100ms，IO 从 5ms 到 1000ms），Dispatchers.IO（适当提高并行度）在所有 CPU 占比 >20% 的场景中均优于虚拟线程。原因是虚拟线程的 CPU 计算部分在载体线程上顺序执行，缺乏 work-stealing 负载均衡；而 Dispatchers.IO 的平台线程池有成熟的 work-stealing 机制。当 CPU 占比 <10%（几乎纯 IO）时，虚拟线程才体现阻塞卸载优势。

5. 实际生产如何选择

这是一个理论值，实际上要结合并发与IO时间观看。

上述资源模型与吞吐量公式给出了理想状态下的边界推导，但在真实业务系统中，并发任务数 NNN 与 IO 阻塞时长 TioT_{\text{io}}Tio​ 并非相互独立的变量：二者共同决定了底层线程池的利用率、内存占用规模与调度开销占比，也直接划定了 Kotlin 协程与 Java 虚拟线程的选型收益边界。脱离并发规模评判IO时长的优劣，或脱离IO时长估算并发承载能力，都会与生产环境的实际表现产生明显偏差。

我们可以从 联合维度象限模型、生产环境修正项、落地选型步骤、典型场景实测 四个层面，把理论公式落地为可执行的判断标准，完整呈现 并发 × IO时长 共同作用下的真实表现。

补充维度：除并发规模和IO时长外，CPU 占比是第三个不应忽视的变量。真实业务中 IO 和 CPU 几乎总是混合的，脱离 CPU 占比谈 IO 调度器选型，与脱离并发规模谈 IO 时长一样，都会得出不完整的结论。实测表明，当 CPU 占比 >20% 时，平台线程池的 work-stealing 优势大于虚拟线程的阻塞卸载优势。

5.1 并发规模 × IO时长 四象限选型模型

我们以「单任务IO阻塞时长」为横轴，「稳态并发任务数」为纵轴，将业务场景划分为四个象限，对应不同的资源表现与最优选型，本质是对 N⋅TioN \cdot T_{\text{io}}N⋅Tio​ 乘积的具象化判断：

该模型的核心判断阈值可由理论公式推导而来：
当 N⋅Tio<200N \cdot T_{\text{io}} < 200N⋅Tio​<200（单位：任务数 × 秒）时，Dispatchers.IO 的平台线程开销完全可控，协程的结构化并发、流式处理能力带来的工程价值更高，虚拟线程的资源收益不足以覆盖改造成本；
当 N⋅Tio>500N \cdot T_{\text{io}} > 500N⋅Tio​>500 时，平台线程池的边际成本开始指数级上升，虚拟线程「阻塞不占用平台线程」的特性开始体现出压倒性的资源优势。

[注5] 实测显示 JDK 21 虚拟线程的创建开销（create+join ~22μs）实际低于协程的 launch+join（~28μs），所谓"虚拟线程创建开销高于协程用户态切换"更多体现在频繁挂载/卸载的内核态开销上，而非创建本身。

[注6] Dispatchers.IO 默认并行度为 max(64, CPU核数)。短IO高并发下，这个限制可能成为吞吐量的主要瓶颈。实测 IO=5ms, N=5,000 时，默认 64 并行度吞吐量为 208K ops/s，而将并行度提升到 1024 后吞吐量可达 434K ops/s（翻倍）。但生产中不应盲目调大此值，需结合 P95 IO 延迟按 Little’s Law 估算合理并行度。详见附录 B。

5.2 生产环境对理论模型的5个关键点

前文的资源公式与吞吐量公式均为理想状态推导，落地到真实JVM与操作系统中，需要补充多个修正因子，否则容易出现理论估算与线上表现的显著偏差。

a. 内存开销：协程并非永远数量级占优

理论上协程动态栈最小仅2KB，虚拟线程固定栈1MB，但真实业务场景中：

• 协程会持有业务对象、上下文引用、回调状态，真实单协程内存占用通常在 10~100KB 区间；
• 虚拟线程的1MB为栈的最大预留值，实际物理内存按使用量提交，浅调用栈的虚拟线程实际内存占用约几十到几百KB。
• 结论：万级并发下协程内存优势明显；十万级以上并发，二者内存差距收窄，协程仍有2~5倍优势，但并非理论上的百倍差距。

实测验证：10,000 虚拟线程（仅 sleep(1ms)）总内存增量约 3 MB（均摊 ~0.3 KB/线程），10,000 协程（仅 delay(1ms)）增量几乎不可测量。在极轻量任务下虚拟线程的实际内存占用远优于理论预期，但协程的 GC 可回收性使其在长时间运⾏中仍占优。

b. 阻塞卸载：虚拟线程并非所有阻塞都能释放

JVM 虚拟线程的「阻塞卸载」仅对 JDK 内置的可中断阻塞API生效（Socket IO、文件IO、sleep、显式锁等），对于第三方 native 阻塞库、JNI调用、未适配的自定义阻塞逻辑，虚拟线程无法卸载载体线程，会直接退化为平台线程阻塞。

• 结论：存在native依赖的长IO场景，虚拟线程收益会打折扣，选型前需验证核心阻塞点是否被JVM适配。

c. 协程延迟：排队延迟是长IO高并发的核心瓶颈

理论延迟公式未考虑线程池饱和后的排队效应，真实场景下需补充排队项：
延迟实际=排队等待时间+Tio线程池大小+Cresume 延迟_{实际} = 排队等待时间 + \frac{T_{\text{io}}}{线程池大小} + C_{\text{resume}}延迟实际​=排队等待时间+线程池大小Tio​​+Cresume​
当并发长IO导致线程池饱和时，排队延迟会成为主导项，吞吐量快速下跌——这也是 Dispatchers.IO 不适合长IO高并发的核心原因，而非IO本身的执行速度。

d. IO时长：长尾分布比均值更关键

理论模型通常使用平均IO时长，但生产环境中IO耗时普遍呈长尾分布，P99耗时往往是均值的3~10倍。这些长尾慢请求会长期占用平台线程，成为线程池膨胀的主要推手。

• 结论：评估 Dispatchers.IO 压力时，应以 P95/P99 IO时长 计算稳态线程数，而非平均耗时，否则会严重低估线程池峰值压力。

e. 调度开销：短IO下虚拟线程切换成本不可忽略

• 协程切换为纯用户态操作，开销约 0.1~1μs，不涉及内核态转换；
• 虚拟线程切换需内核态挂载/卸载，开销约 1~10μs，与平台线程上下文切换同量级。
  当 Tio<1msT_{\text{io}} < 1\text{ms}Tio​<1ms 时，调度开销占比会显著提升，此时虚拟线程的吞吐量反而会低于协程线程池，这也呼应了前文 Tio>10msT_{\text{io}} > 10\text{ms}Tio​>10ms 时虚拟线程才体现优势的临界点。

实测验证：IO=1ms, N=10,000 时，IO(默认) 吞吐量 1,113,728 ops/s，原生 VT 仅 635,467 ops/s（IO 领先 75%），充分验证了极短 IO 下虚拟线程调度开销占比过高的结论。但到 IO=5ms 时，差距缩小至 6%（685K vs 644K），说明临界区间在 1~5ms 之间。

三、落地选型：三步判断法（可直接复用）

面对真实业务场景，可按以下步骤快速判断选型，无需复杂公式推导：

第一步：量化稳态平台线程需求

按业务峰值的 P95 IO 耗时，计算使用 Dispatchers.IO 时的稳态活跃线程数：
预估活跃线程数=峰值并发任务数×P95 IO耗时(ms)1000 \text{预估活跃线程数} = \text{峰值并发任务数} \times \frac{\text{P95 IO耗时(ms)}}{1000}预估活跃线程数=峰值并发任务数×1000P95 IO耗时(ms)​

• 结果 < 100：轻量负载，维持现有协程方案即可；
• 结果 100~500：中负载，可监控观察，配合信号量限流兜底；
• 结果 > 500：重负载，Dispatchers.IO 已接近瓶颈，优先考虑虚拟线程。

补充：结合 Dispatchers.IO 默认并行度（max(64, CPU核数)）评估。若预估活跃线程数超过默认并行度，实际吞吐量将受限于线程池排队，建议显式调高 limitedParallelism(N) 或切换到虚拟线程。推荐并行度公式：建议并行度 = max(64, 预估活跃线程数 × 1.2)。但不建议直接设为无限制（Int.MAX_VALUE），实测无限制反而因过多线程竞争导致性能下降。

第二步：匹配业务技术生态

• 若业务深度依赖协程生态（Flow、Channel、结构化并发、超时取消、多任务组合），优先保留协程体系，通过切换调度器（虚拟线程转协程调度器）解决长IO问题，而非完全抛弃协程改用原生虚拟线程；
• 若业务为纯批量阻塞任务、无复杂异步编排、以Java代码为主，可直接使用原生虚拟线程 Executor，改造成本最低。

第三步：压测验证核心指标

任何理论估算都需要压测兜底，重点验证三个维度：

1. 活跃平台线程数：是否在预期范围内，是否出现无上限持续增长；
2. GC表现：虚拟线程的对象创建、协程的上下文对象是否带来额外GC压力；
3. 吞吐量与长尾延迟：对比相同资源下的QPS与P99延迟，验证收益是否符合预期。
4. （新增）混合负载下的 CPU 占比：若 CPU 计算耗时占总耗时 >20%，建议在压测中额外对比 Dispatchers.IO（适当提高并行度）与虚拟线程的表现——实测中平台线程池的 work-stealing 在此场景下具有显著优势。

5.4 典型业务场景的实测参考

基于常规服务器配置（8核16G，JDK 21，Kotlin 2.0），三组典型场景的实测对比可作为选型参考：

1. Redis短查询：IO平均5ms，峰值并发5000
   • Dispatchers.IO：稳态活跃线程约25条，吞吐量4.8万 ops/s[注7]，GC开销可忽略；
   • 虚拟线程调度器：吞吐量4.2万 ops/s，调度开销占比提升，GC频率略高；
   • 结论：短IO高并发下，协程平台线程池更优。
2. 第三方HTTP接口调用：IO平均200ms，峰值并发2000
   • Dispatchers.IO：稳态活跃线程约400条，内存占用明显增加，吞吐量约0.9万 ops/s；
   • 虚拟线程调度器：载体线程稳定在16条，内存节省60%，吞吐量约1.0万 ops/s；
   • 结论：中长IO中高并发下，虚拟线程开始体现资源优势。
3. 慢SQL批量导出：IO平均3s，峰值并发1000
   • Dispatchers.IO：稳态线程逼近1000条，系统线程数告警，上下文切换飙升，吞吐量约300 ops/s，随时可能雪崩；
   • 虚拟线程调度器：载体线程稳定在20条以内，内存仅为平台线程方案的1/10，吞吐量约320 ops/s，运行平稳；
   • 结论：长IO高并发下，虚拟线程方案具备压倒性优势。

[注7] 该数据基于含真实 Redis 网络 IO 的生产场景（含序列化/反序列化/网络往返）。在纯 delay(5ms) 模拟测试中，IO(默认) 可达 512K ops/s，远高于此值。这是因为真实 Redis IO 除了网络延迟外还包含连接池竞争、序列化开销等。使用纯 delay 的基准测试只能反映调度器本身的差异趋势，不能直接作为生产容量规划的绝对值。

实测验证（纯 delay 模拟，12核 JDK 21）：场景1（IO=5ms, N=5,000）IO(默认) 512K vs 原生VT 393K，确认协程更优的趋势；场景2（IO=200ms, N=2,000）IO(默认) 23.6K vs 原生VT 24.2K，确认虚拟线程略优的趋势；场景3（IO=3000ms, N=1,000）二者均接近 330 ops/s（IO 耗时完全主导），虚拟线程在内存可控性上的优势成为关键区分点。

5. 结论

协程与虚拟线程并非对立关系，在真实系统中往往是混合共存的：CPU密集型计算、短IO快调用走 Dispatchers.Default 或 Dispatchers.IO，利用平台线程池的低调度开销；长阻塞IO、高并发外部调用走 Dispatchers.Virtual（需自己实现），利用虚拟线程的阻塞卸载能力；全链路统一在协程的结构化并发体系下管理，兼顾工程效率与资源效率。

理论公式帮我们理解底层资源边界，而「并发 × 时长」的联合判断，才是生产环境选型的核心依据。脱离并发谈IO时长、脱离时长谈并发规模，都无法得出准确的技术选型结论。同样地——实测新增的发现表明——脱离 CPU 占比谈调度器选型也会失之偏颇：当业务逻辑中 CPU 计算占比超过 20% 时，平台线程池的 work-stealing 机制往往比虚拟线程的阻塞卸载带来更大的吞吐量收益，这一维度值得在生产选型中额外关注。

虚拟线程适用于长 IO 大并发系统，但对于内存限制或短 IO 场景、以及 CPU 占比较高的混合负载场景，Kotlin IO 协程凭借轻量和高效调度更具优势。建议在关键系统中混合两种策略（如主逻辑用协程、外部接口调用用虚拟线程），以平衡响应时间与资源利用率。

附录 A：IO 密集型实测矩阵

环境：JDK 21.0.5 · Kotlin 2.3.21 · 12核CPU · BenchmarkVerification.kt
数值单位：吞吐量 ops/s

IO=1ms（极短IO）：

IO=5ms（短IO）：

IO=50ms（中IO）：

IO=200ms（长IO）：

IO=1000ms（超长IO）： 所有调度器吞吐量趋同于 N/1.0sN / 1.0\text{s}N/1.0s。

附录 B：Dispatchers.IO 默认并行度限制的梯度影响

IO=5ms, N=5,000

当 IO 延迟达到 200ms 时，所有并行度限制下的吞吐量趋同（约 22,000~23,000 ops/s）——IO 耗时完全主导，并行度限制不再产生影响。

附录 C：调度创建开销实测

附录 D：混合 CPU+IO 负载实测

附录 E：测试代码

1import kotlinx.coroutines.*
2import java.util.concurrent.ConcurrentLinkedQueue
3import java.util.concurrent.Executors
4import java.util.concurrent.atomic.AtomicLong
5import kotlin.math.sqrt
6import kotlin.system.measureNanoTime
7import kotlin.system.measureTimeMillis
8
9/**
10 * 虚拟线程 vs 平台线程 vs 协程 全面性能对比基准测试
11 *
12 * 测试三个调度器:
13 *   1. Platform  - 传统平台线程 (Thread)
14 *   2. Virtual   - Java 虚拟线程 (Thread.ofVirtual)
15 *   3. IO        - 协程 Dispatchers.IO (平台线程池)
16 *   4. IO(无限制) - 协程 Dispatchers.IO.limitedParallelism(Int.MAX_VALUE) 解除线程数限制
17 *   5. Virtual-Dispatcher - 协程 Dispatchers.Virtual (虚拟线程调度器)
18 *
19 * @author : zimo
20 * @date : 2026/06/24
21 */
22
23// ============================================================
24// 数据结构
25// ============================================================
26
27data class PerfResult(
28    val testSuite: String,
29    val testName: String,
30    val dispatcher: String,
31    val concurrency: Int,
32    val workMs: Long,
33    val totalOps: Long,
34    val durationMs: Long,
35    val throughput: Double,
36    val avgLatencyMs: Double,
37    val p99LatencyMs: Double,
38    val memoryDeltaMB: Double,
39)
40
41val allResults = mutableListOf<PerfResult>()
42
43// ============================================================
44// 工具函数
45// ============================================================
46
47fun String.times(n: Int) = this.repeat(n)
48
49/** 测量内存增量 */
50inline fun <T> measureMemoryDelta(block: () -> T): Pair<T, Double> {
51    System.gc()
52    Thread.sleep(300)
53    val before = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()
54    val result = block()
55    System.gc()
56    Thread.sleep(300)
57    val after = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()
58    return result to ((after - before).toDouble() / (1024 * 1024))
59}
60
61/** CPU 密集型工作: 计算第 N 个素数 */
62fun cpuWork(ms: Long) {
63    val start = System.nanoTime()
64    var n = 2L
65    while ((System.nanoTime() - start) / 1_000_000 < ms) {
66        var prime = true
67        val limit = sqrt(n.toDouble()).toLong() + 1
68        for (i in 2..limit) {
69            if (n % i == 0L) { prime = false; break }
70        }
71        @Suppress("UNUSED_EXPRESSION")
72        prime // busy work result discarded
73        n++
74    }
75}
76
77/** 获取当前活跃线程数估算 */
78fun activeThreadEstimate(): Int = Thread.activeCount()
79
80// ============================================================
81// 测试 1: 纯 CPU 密集型
82// ============================================================
83
84suspend fun benchmarkCpuIntensive() {
85    println("=".times(60))
86    println("测试 1: 纯 CPU 密集型 (计算素数)")
87    println("=".times(60))
88
89    val cpuCores = Runtime.getRuntime().availableProcessors()
90    val workMs = 50L
91    val concurrencyLevels = listOf(cpuCores, cpuCores * 2, cpuCores * 4, cpuCores * 10, 200)
92
93    // CPU 密集型只用 Default / Virtual-Dispatcher / 原生 VT
94    // IO 调度器不适合 CPU 密集任务
95    val dispatchers = listOf(
96        "Default" to Dispatchers.Default,
97    )
98
99    val vtDispatcher = Executors.newVirtualThreadPerTaskExecutor().asCoroutineDispatcher()
100
101    println("%-8s | %-12s | %12s | %12s | %12s".format("并发", "调度器", "吞吐量ops/s", "平均延迟ms", "P99延迟ms"))
102    println("-".times(70))
103
104    for (n in concurrencyLevels) {
105        for ((name, dispatcher) in dispatchers) {
106            val result = runCoroutineCpuTest("CPU密集", "素数计算${workMs}ms", name, n, workMs, dispatcher)
107            allResults.add(result)
108            println("%-8d | %-12s | %12.0f | %12.1f | %12.1f".format(
109                n, name, result.throughput, result.avgLatencyMs, result.p99LatencyMs
110            ))
111        }
112
113        // 原生虚拟线程 (非协程)
114        val vtResult = runNativeVTCpuTest("CPU密集", "素数计算${workMs}ms", n, workMs)
115        allResults.add(vtResult)
116        println("%-8d | %-12s | %12.0f | %12.1f | %12.1f".format(
117            n, "原生VT", vtResult.throughput, vtResult.avgLatencyMs, vtResult.p99LatencyMs
118        ))
119    }
120    vtDispatcher.close()
121    println()
122}
123
124/** 协程 CPU 测试 */
125suspend fun runCoroutineCpuTest(suite: String, name: String, dispatcherLabel: String, n: Int, workMs: Long, dispatcher: CoroutineDispatcher): PerfResult {
126    val latencies = ConcurrentLinkedQueue<Long>()
127    val ops = AtomicLong(0)
128    val rounds = 3
129
130    // 预热
131    coroutineScope { (1..n).map { launch(dispatcher) { cpuWork(workMs / 2) } }.joinAll() }
132
133    val (_, memDelta) = measureMemoryDelta {
134        runBlocking {
135            val duration = measureTimeMillis {
136                repeat(rounds) {
137                    coroutineScope {
138                        (1..n).map {
139                            launch(dispatcher) {
140                                val t0 = System.nanoTime()
141                                cpuWork(workMs)
142                                latencies.add((System.nanoTime() - t0) / 1_000_000)
143                                ops.incrementAndGet()
144                            }
145                        }.joinAll()
146                    }
147                }
148            }
149        }
150    }
151
152    val sorted = latencies.toList().sorted()
153    val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
154    val avg = latencies.toList().average()
155
156    return PerfResult(suite, name, dispatcherLabel, n, workMs, ops.get(), 0, ops.get().toDouble(), avg, p99, memDelta)
157}
158
159/** 原生虚拟线程 CPU 测试 */
160fun runNativeVTCpuTest(suite: String, name: String, n: Int, workMs: Long): PerfResult {
161    val latencies = ConcurrentLinkedQueue<Long>()
162    val ops = AtomicLong(0)
163    val rounds = 3
164
165    val executor = Executors.newVirtualThreadPerTaskExecutor()
166    // 预热
167    (1..n).map { executor.submit { cpuWork(workMs / 2) } }.forEach { it.get() }
168
169    val (_, memDelta) = measureMemoryDelta {
170        val duration = measureTimeMillis {
171            repeat(rounds) {
172                val futures = (1..n).map {
173                    executor.submit {
174                        val t0 = System.nanoTime()
175                        cpuWork(workMs)
176                        latencies.add((System.nanoTime() - t0) / 1_000_000)
177                        ops.incrementAndGet()
178                    }
179                }
180                futures.forEach { it.get() }
181            }
182        }
183    }
184    executor.close()
185
186    val sorted = latencies.toList().sorted()
187    val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
188    val avg = latencies.toList().average()
189
190    return PerfResult(suite, name, "原生VT", n, workMs, ops.get(), 0, ops.get().toDouble(), avg, p99, memDelta)
191}
192
193// ============================================================
194// 测试 2: 纯 IO 密集型 (delay 模拟)
195// ============================================================
196
197suspend fun benchmarkIOIntensive() {
198    println("=".times(60))
199    println("测试 2: 纯 IO 密集型 (delay 模拟阻塞)")
200    println("=".times(60))
201
202    // IO 时长 × 并发 矩阵测试
203    val ioDelays = listOf(1L, 5L, 50L, 200L, 1000L)   // ms
204    val concurrencyLevels = listOf(100, 500, 1000, 5000, 10000)
205
206    // 调度器: IO(默认), IO(无限制), Virtual-Dispatcher, 原生VT
207    println()
208    println("=== 矩阵测试: 不同 IO 时长 × 不同并发 ===")
209
210    for (delayMs in ioDelays) {
211        println("\n--- IO延迟 = ${delayMs}ms ---")
212        println("%-8s | %-16s | %12s | %12s | %18s".format("N", "调度器", "吞吐量ops/s", "均延迟ms", "内存增量MB"))
213        println("-".times(80))
214
215        for (n in concurrencyLevels) {
216            // Dispatchers.IO (默认)
217            val rIO = runIOTest("IO密集", "delay=${delayMs}ms", "IO(默认)", n, delayMs, Dispatchers.IO)
218            allResults.add(rIO)
219            println("%-8d | %-16s | %12.0f | %12.1f | %18.2f".format(n, "IO(默认)", rIO.throughput, rIO.avgLatencyMs, rIO.memoryDeltaMB))
220
221            // Dispatchers.IO (无限制)
222            val ioUnlimited = Dispatchers.IO.limitedParallelism(Int.MAX_VALUE)
223            val rIOU = runIOTest("IO密集", "delay=${delayMs}ms", "IO(无限制)", n, delayMs, ioUnlimited)
224            allResults.add(rIOU)
225            println("%-8d | %-16s | %12.0f | %12.1f | %18.2f".format(n, "IO(无限制)", rIOU.throughput, rIOU.avgLatencyMs, rIOU.memoryDeltaMB))
226
227            // 跳过小并发的 VT (没意义)
228            if (n >= 500) {
229                val vtDispatcher = Executors.newVirtualThreadPerTaskExecutor().asCoroutineDispatcher()
230                val rVD = runIOTest("IO密集", "delay=${delayMs}ms", "VT-Dispatcher", n, delayMs, vtDispatcher)
231                allResults.add(rVD)
232                println("%-8d | %-16s | %12.0f | %12.1f | %18.2f".format(n, "VT-Dispatcher", rVD.throughput, rVD.avgLatencyMs, rVD.memoryDeltaMB))
233                vtDispatcher.close()
234
235                val rVT = runNativeVTTest("IO密集", "delay=${delayMs}ms", n, delayMs)
236                allResults.add(rVT)
237                println("%-8d | %-16s | %12.0f | %12.1f | %18.2f".format(n, "原生VT", rVT.throughput, rVT.avgLatencyMs, rVT.memoryDeltaMB))
238            }
239        }
240    }
241    println()
242}
243
244/** 协程 IO 测试 */
245suspend fun runIOTest(suite: String, name: String, dispLabel: String, n: Int, delayMs: Long, dispatcher: CoroutineDispatcher): PerfResult {
246    val latencies = ConcurrentLinkedQueue<Long>()
247    val rounds = if (n > 2000) 2 else 5
248    var totalOps = 0L
249
250    // 预热
251    coroutineScope { (1..n).map { launch(dispatcher) { delay(delayMs) } }.joinAll() }
252
253    val (_, memDelta) = measureMemoryDelta {
254        runBlocking {
255            repeat(rounds) {
256                coroutineScope {
257                    val jobs = (1..n).map {
258                        launch(dispatcher) {
259                            val t0 = System.nanoTime()
260                            delay(delayMs)
261                            latencies.add((System.nanoTime() - t0) / 1_000_000)
262                        }
263                    }
264                    totalOps += jobs.size
265                    jobs.joinAll()
266                }
267            }
268        }
269    }
270
271    val sorted = latencies.toList().sorted()
272    val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
273    val avg = latencies.toList().average()
274    val throughput = if (avg > 0) 1000.0 / (avg / rounds) * n else 0.0
275    // 更好的吞吐量计算: 在理想情况下，吞吐量 = N / (IO延迟) × (调度器并行度系数)
276    // 我们用实际: ops / (预估总耗时) 估算
277    val estimatedThroughput = if (sorted.isNotEmpty()) {
278        totalOps * 1000.0 / (sorted.sum() / sorted.size * rounds)
279    } else 0.0
280    // 简化: 吞吐量 = 并发数 / (延迟/rounds) = n * rounds / avg_latency_in_seconds
281    val estimatedDurationPerRound = sorted.map { it.toDouble() }.average()
282    val simpleThroughput = if (estimatedDurationPerRound > 0) n * 1000.0 / estimatedDurationPerRound else 0.0
283
284    return PerfResult(suite, name, dispLabel, n, delayMs, totalOps, 0, simpleThroughput, avg, p99, memDelta)
285}
286
287/** 原生虚拟线程 IO 测试 */
288fun runNativeVTTest(suite: String, name: String, n: Int, delayMs: Long): PerfResult {
289    val latencies = ConcurrentLinkedQueue<Long>()
290    val rounds = if (n > 2000) 2 else 5
291    var totalOps = 0L
292
293    val executor = Executors.newVirtualThreadPerTaskExecutor()
294    // 预热
295    (1..n).map { executor.submit { Thread.sleep(delayMs) } }.forEach { it.get() }
296
297    val (_, memDelta) = measureMemoryDelta {
298        repeat(rounds) {
299            val futures = (1..n).map {
300                executor.submit {
301                    val t0 = System.nanoTime()
302                    Thread.sleep(delayMs)
303                    latencies.add((System.nanoTime() - t0) / 1_000_000)
304                }
305            }
306            totalOps += futures.size
307            futures.forEach { it.get() }
308        }
309    }
310    executor.close()
311
312    val sorted = latencies.toList().sorted()
313    val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
314    val avg = latencies.toList().average()
315    val simpleThroughput = if (avg > 0) n * 1000.0 / avg else 0.0
316
317    return PerfResult(suite, name, "原生VT", n, delayMs, totalOps, 0, simpleThroughput, avg, p99, memDelta)
318}
319
320// ============================================================
321// 测试 3: 混合 CPU + IO (贴近真实业务)
322// ============================================================
323
324suspend fun benchmarkMixed() {
325    println("=".times(60))
326    println("测试 3: 混合 CPU+IO (模拟真实业务)")
327    println("=".times(60))
328
329    // 场景: 50% CPU计算 + 50% IO等待 (如: 处理请求 + 调外部服务 + 返回结果)
330    data class MixedScenario(val label: String, val cpuMs: Long, val ioMs: Long, val concurrency: Int)
331
332    val scenarios = listOf(
333        MixedScenario("轻量API", 2, 5, 500),
334        MixedScenario("普通业务", 10, 20, 1000),
335        MixedScenario("数据处理+外部调用", 20, 100, 2000),
336        MixedScenario("复杂计算+慢IO", 50, 500, 1000),
337        MixedScenario("报表生成+批量IO", 100, 1000, 500),
338    )
339
340    println()
341    println("%-20s | %5s/%5s | %6s | %-16s | %10s | %10s | %10s".format(
342        "场景", "CPUms", "IOms", "N", "调度器", "吞吐量", "均延迟ms", "内存MB"
343    ))
344    println("-".times(100))
345
346    for (scenario in scenarios) {
347        val cpuMs = scenario.cpuMs
348        val ioMs = scenario.ioMs
349        val n = scenario.concurrency
350
351        // IO (默认)
352        val r1 = runMixedTest("混合负载", scenario.label, "IO(默认)", n, cpuMs, ioMs, Dispatchers.IO)
353        allResults.add(r1)
354        println("%-20s | %5d/%5d | %6d | %-16s | %10.0f | %10.1f | %10.2f".format(
355            scenario.label, cpuMs, ioMs, n, "IO(默认)", r1.throughput, r1.avgLatencyMs, r1.memoryDeltaMB
356        ))
357
358        // IO (无限制)
359        val ioUnlimited = Dispatchers.IO.limitedParallelism(Int.MAX_VALUE)
360        val r2 = runMixedTest("混合负载", scenario.label, "IO(无限制)", n, cpuMs, ioMs, ioUnlimited)
361        allResults.add(r2)
362        println("%-20s | %5d/%5d | %6d | %-16s | %10.0f | %10.1f | %10.2f".format(
363            scenario.label, cpuMs, ioMs, n, "IO(无限制)", r2.throughput, r2.avgLatencyMs, r2.memoryDeltaMB
364        ))
365
366        // VT-Dispatcher
367        val vtDispatcher = Executors.newVirtualThreadPerTaskExecutor().asCoroutineDispatcher()
368        val r3 = runMixedTest("混合负载", scenario.label, "VT-Dispatcher", n, cpuMs, ioMs, vtDispatcher)
369        allResults.add(r3)
370        println("%-20s | %5d/%5d | %6d | %-16s | %10.0f | %10.1f | %10.2f".format(
371            scenario.label, cpuMs, ioMs, n, "VT-Dispatcher", r3.throughput, r3.avgLatencyMs, r3.memoryDeltaMB
372        ))
373        vtDispatcher.close()
374
375        // 原生 VT
376        val r4 = runNativeVTMixedTest("混合负载", scenario.label, n, cpuMs, ioMs)
377        allResults.add(r4)
378        println("%-20s | %5d/%5d | %6d | %-16s | %10.0f | %10.1f | %10.2f".format(
379            scenario.label, cpuMs, ioMs, n, "原生VT", r4.throughput, r4.avgLatencyMs, r4.memoryDeltaMB
380        ))
381    }
382    println()
383}
384
385/** 协程混合负载测试: CPU计算 + delay(IO) 交替 */
386suspend fun runMixedTest(suite: String, name: String, dispLabel: String, n: Int, cpuMs: Long, ioMs: Long, dispatcher: CoroutineDispatcher): PerfResult {
387    val latencies = ConcurrentLinkedQueue<Long>()
388    val rounds = if (n > 1000) 3 else 5
389    var totalOps = 0L
390
391    // 预热
392    coroutineScope {
393        (1..n).map {
394            launch(dispatcher) {
395                cpuWork(cpuMs / 2)
396                delay(ioMs / 2)
397            }
398        }.joinAll()
399    }
400
401    val (_, memDelta) = measureMemoryDelta {
402        runBlocking {
403            repeat(rounds) {
404                coroutineScope {
405                    val jobs = (1..n).map {
406                        launch(dispatcher) {
407                            val t0 = System.nanoTime()
408                            // 混合工作: CPU计算 → IO等待 → CPU计算
409                            cpuWork(cpuMs)
410                            delay(ioMs)
411                            cpuWork(cpuMs / 3)
412                            latencies.add((System.nanoTime() - t0) / 1_000_000)
413                        }
414                    }
415                    totalOps += jobs.size
416                    jobs.joinAll()
417                }
418            }
419        }
420    }
421
422    val sorted = latencies.toList().sorted()
423    val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
424    val avg = latencies.toList().average()
425    val throughput = if (avg > 0) n * 1000.0 / avg else 0.0
426
427    return PerfResult(suite, name, dispLabel, n, cpuMs + ioMs, totalOps, 0, throughput, avg, p99, memDelta)
428}
429
430/** 原生虚拟线程混合负载测试 */
431fun runNativeVTMixedTest(suite: String, name: String, n: Int, cpuMs: Long, ioMs: Long): PerfResult {
432    val latencies = ConcurrentLinkedQueue<Long>()
433    val rounds = if (n > 1000) 3 else 5
434    var totalOps = 0L
435
436    val executor = Executors.newVirtualThreadPerTaskExecutor()
437    // 预热
438    (1..n).map {
439        executor.submit {
440            cpuWork(cpuMs / 2)
441            Thread.sleep(ioMs / 2)
442        }
443    }.forEach { it.get() }
444
445    val (_, memDelta) = measureMemoryDelta {
446        repeat(rounds) {
447            val futures = (1..n).map {
448                executor.submit {
449                    val t0 = System.nanoTime()
450                    cpuWork(cpuMs)
451                    Thread.sleep(ioMs)
452                    cpuWork(cpuMs / 3)
453                    latencies.add((System.nanoTime() - t0) / 1_000_000)
454                }
455            }
456            totalOps += futures.size
457            futures.forEach { it.get() }
458        }
459    }
460    executor.close()
461
462    val sorted = latencies.toList().sorted()
463    val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
464    val avg = latencies.toList().average()
465    val throughput = if (avg > 0) n * 1000.0 / avg else 0.0
466
467    return PerfResult(suite, name, "原生VT", n, cpuMs + ioMs, totalOps, 0, throughput, avg, p99, memDelta)
468}
469
470// ============================================================
471// 测试 4: 内存开销对比
472// ============================================================
473
474suspend fun benchmarkMemoryOverhead() {
475    println("=".times(60))
476    println("测试 4: 内存开销对比 (大量并发任务)")
477    println("=".times(60))
478
479    val levels = listOf(100, 1000, 5000, 10000)
480
481    println("%-8s | %-16s | %18s | %18s".format("任务数", "调度器", "内存增量MB", "单任务估算KB"))
482    println("-".times(70))
483
484    for (count in levels) {
485        // 协程 IO
486        val (_, coMem) = measureMemoryDelta {
487            runBlocking {
488                coroutineScope {
489                    (1..count).map { launch(Dispatchers.IO) { delay(1) } }.joinAll()
490                }
491            }
492        }
493        println("%-8d | %-16s | %18.2f | %18.2f".format(count, "IO协程", coMem, coMem * 1024 / count))
494
495        // 协程 Default
496        val (_, defMem) = measureMemoryDelta {
497            runBlocking {
498                coroutineScope {
499                    (1..count).map { launch(Dispatchers.Default) { delay(1) } }.joinAll()
500                }
501            }
502        }
503        println("%-8d | %-16s | %18.2f | %18.2f".format(count, "Default协程", defMem, defMem * 1024 / count))
504
505        // 虚拟线程 (原生)
506        val (_, vtMem) = measureMemoryDelta {
507            val executor = Executors.newVirtualThreadPerTaskExecutor()
508            val futures = (1..count).map { executor.submit { Thread.sleep(1) } }
509            futures.forEach { it.get() }
510            executor.close()
511        }
512        println("%-8d | %-16s | %18.2f | %18.2f".format(count, "虚拟线程", vtMem, vtMem * 1024 / count))
513
514        // 平台线程 (少量，因为开销大)
515        if (count <= 1000) {
516            val (_, ptMem) = measureMemoryDelta {
517                val executor = Executors.newCachedThreadPool()
518                val futures = (1..count).map { executor.submit { Thread.sleep(1) } }
519                futures.forEach { it.get() }
520                executor.shutdown()
521            }
522            println("%-8d | %-16s | %18.2f | %18.2f".format(count, "平台线程池", ptMem, ptMem * 1024 / count))
523        } else {
524            println("%-8d | %-16s | %18s | %18s".format(count, "平台线程池", "(跳过-资源过大)", "-"))
525        }
526    }
527    println()
528}
529
530// ============================================================
531// 测试 5: 调度/创建开销
532// ============================================================
533
534fun benchmarkSchedulingOverhead() {
535    println("=".times(60))
536    println("测试 5: 任务创建与调度开销")
537    println("=".times(60))
538
539    val iterations = 10_000
540    val warmup = 1000
541
542    data class OverheadResult(val label: String, val totalNs: Long, val count: Int) {
543        val avgNs: Double get() = totalNs.toDouble() / count
544    }
545
546    val results = mutableListOf<OverheadResult>()
547
548    // 1. 协程 launch + join (最小开销)
549    runBlocking {
550        repeat(warmup) { launch(Dispatchers.Default) { }.join() }
551        val t = measureNanoTime {
552            repeat(iterations) { launch(Dispatchers.Default) { }.join() }
553        }
554        results.add(OverheadResult("协程launch+join(Default)", t, iterations))
555    }
556
557    // 2. 协程 launch + delay(0) + join
558    runBlocking {
559        repeat(warmup / 10) { launch(Dispatchers.IO) { delay(0) }.join() }
560        val t = measureNanoTime {
561            repeat(iterations / 10) { launch(Dispatchers.IO) { delay(0) }.join() }
562        }
563        results.add(OverheadResult("协程delay(0)+IO调度", t, iterations / 10))
564    }
565
566    // 3. 原生虚拟线程 create+start+join
567    repeat(warmup) { Thread.ofVirtual().start { }.join() }
568    val vtCreate = measureNanoTime {
569        repeat(iterations) { Thread.ofVirtual().start { }.join() }
570    }
571    results.add(OverheadResult("虚拟线程create+join", vtCreate, iterations))
572
573    // 4. 虚拟线程 Executor submit+get
574    val vtExecutor = Executors.newVirtualThreadPerTaskExecutor()
575    repeat(warmup) { vtExecutor.submit { }.get() }
576    val vtSubmit = measureNanoTime {
577        repeat(iterations) { vtExecutor.submit { }.get() }
578    }
579    results.add(OverheadResult("虚拟线程submit+get", vtSubmit, iterations))
580    vtExecutor.close()
581
582    // 5. 平台线程 create+start+join
583    repeat(warmup / 100) { val t = Thread { }; t.start(); t.join() }
584    val ptCreate = measureNanoTime {
585        repeat(iterations / 100) { val t = Thread { }; t.start(); t.join() }
586    }
587    results.add(OverheadResult("平台线程create+join", ptCreate, iterations / 100))
588
589    // 6. 平台线程池 submit+get
590    val ptExecutor = Executors.newCachedThreadPool()
591    repeat(warmup / 10) { ptExecutor.submit { }.get() }
592    val ptSubmit = measureNanoTime {
593        repeat(iterations / 10) { ptExecutor.submit { }.get() }
594    }
595    results.add(OverheadResult("平台线程池submit+get", ptSubmit, iterations / 10))
596    ptExecutor.shutdown()
597
598    println("%-30s | %10s | %12s".format("操作", "平均延迟", "相对比"))
599    println("-".times(60))
600    val baseline = results.first().avgNs
601    for (r in results) {
602        val ratio = r.avgNs / baseline
603        val unit = if (r.avgNs > 10_000) "μs" else "ns"
604        val display = if (r.avgNs > 10_000) r.avgNs / 1000 else r.avgNs
605        println("%-30s | %8.1f %-3s | %10.1fx".format(r.label, display, unit, ratio))
606    }
607    println()
608}
609
610// ============================================================
611// 测试 6: 线程池并发限制影响分析
612// ============================================================
613
614suspend fun benchmarkThreadPoolLimits() {
615    println("=".times(60))
616    println("测试 6: Dispatchers.IO 线程池限制影响分析")
617    println("=".times(60))
618    println("  说明: Dispatchers.IO 默认并行度 = max(64, CPU核数)")
619    println("  本测试对比不同并行度限制下的表现")
620    println()
621
622    val ioDelays = listOf(5L, 50L, 200L) // ms
623    val concurrency = 5000
624
625    val limits = listOf(64, 128, 256, 512, 1024, Int.MAX_VALUE)
626
627    println("%-6s | %-12s | %12s | %12s | %12s".format("IOms", "并行度限制", "吞吐量ops/s", "均延迟ms", "P99延迟ms"))
628    println("-".times(70))
629
630    for (delayMs in ioDelays) {
631        for (limit in limits) {
632            val dispatcher = Dispatchers.IO.limitedParallelism(limit)
633            val label = if (limit == Int.MAX_VALUE) "无限制" else "$limit"
634
635            val latencies = ConcurrentLinkedQueue<Long>()
636            // 预热
637            coroutineScope { (1..concurrency).map { launch(dispatcher) { delay(delayMs) } }.joinAll() }
638
639            coroutineScope {
640                val jobs = (1..concurrency).map {
641                    launch(dispatcher) {
642                        val t0 = System.nanoTime()
643                        delay(delayMs)
644                        latencies.add((System.nanoTime() - t0) / 1_000_000)
645                    }
646                }
647                jobs.joinAll()
648            }
649
650            val sorted = latencies.toList().sorted()
651            val p99 = if (sorted.isNotEmpty()) sorted[(sorted.size * 0.99).toInt().coerceAtMost(sorted.size - 1)].toDouble() else 0.0
652            val avg = latencies.toList().average()
653            val throughput = if (avg > 0) concurrency * 1000.0 / avg else 0.0
654
655            println("%-6d | %-12s | %12.0f | %12.1f | %12.1f".format(delayMs, label, throughput, avg, p99))
656        }
657        println()
658    }
659    println()
660}
661
662// ============================================================
663// 生成测试报告
664// ============================================================
665
666fun generateReport() {
667    val sep = "=".times(60)
668    println(sep)
669    println("测试报告")
670    println(sep)
671
672    // === 按测试套汇总 ===
673    val suites = allResults.groupBy { it.testSuite }
674    for ((suite, results) in suites) {
675        println("\n## $suite")
676        println()
677
678        val byDispatcher = results.groupBy { it.dispatcher }
679        val dispatchers = byDispatcher.keys.sorted()
680
681        // 表头
682        print("%-20s".format("测试"))
683        for (d in dispatchers) {
684            print(" | %12s".format(d.take(12)))
685        }
686        println()
687        print("-".times(20))
688        for (d in dispatchers) print(" | ${"-".times(12)}")
689        println()
690
691        val byTest = results.groupBy { "${it.concurrency}@${it.workMs}ms" to it.testName }
692        for ((key, group) in byTest) {
693            print("%-20s".format(group.first().testName.take(20)))
694            for (d in dispatchers) {
695                val r = group.find { it.dispatcher == d }
696                if (r != null) {
697                    print(" | %12.0f".format(r.throughput))
698                } else {
699                    print(" | %12s".format("-"))
700                }
701            }
702            println()
703        }
704    }
705
706    // === 综合排名 ===
707    println("\n## 综合排名 (按场景-吞吐量)")
708    println()
709    val ranked = allResults.sortedByDescending { it.throughput }
710    println("%-4s | %-20s | %-16s | %8s | %8s | %10s | %10s".format(
711        "排名", "场景", "调度器", "并发", "工作ms", "吞吐量", "延迟ms"
712    ))
713    println("-".times(85))
714    ranked.take(30).forEachIndexed { idx, r ->
715        println("%-4d | %-20s | %-16s | %8d | %8d | %10.0f | %10.1f".format(
716            idx + 1, r.testName.take(20), r.dispatcher, r.concurrency, r.workMs, r.throughput, r.avgLatencyMs
717        ))
718    }
719
720    // === 调度器全局对比 ===
721    println("\n## 调度器全局对比 (所有场景的调和均值)")
722    println()
723    val globalByDispatcher = allResults.groupBy { it.dispatcher }.mapValues { (_, list) ->
724        val valid = list.filter { it.throughput > 0 }
725        if (valid.isNotEmpty()) valid.map { it.throughput }.average() else 0.0
726    }
727    globalByDispatcher.entries.sortedByDescending { it.value }.forEach { (disp, avg) ->
728        println("  $disp: 平均吞吐量 ${"%.0f".format(avg)} ops/s")
729    }
730}
731
732// ============================================================
733// 主入口
734// ============================================================
735
736fun main() = runBlocking {
737    val cpuCores = Runtime.getRuntime().availableProcessors()
738    val javaVersion = System.getProperty("java.version")
739
740    println("╔══════════════════════════════════════════════════════════════╗")
741    println("║  虚拟线程 / 协程 / 平台线程 全面性能对比基准测试               ║")
742    println("║  JDK: $javaVersion  |  CPU: ${cpuCores}核  |  OS: ${System.getProperty("os.name")} ║")
743    println("╚══════════════════════════════════════════════════════════════╝")
744    println()
745    println("测试调度器:")
746    println("  - IO(默认):     Dispatchers.IO (max(64, CPU核)并行)")
747    println("  - IO(无限制):   Dispatchers.IO.limitedParallelism(MAX)")
748    println("  - Default:      Dispatchers.Default (CPU核数个线程)")
749    println("  - VT-Dispatcher: Dispatchers.Virtual (虚拟线程调度器)")
750    println("  - 原生VT:        Thread.ofVirtual() (非协程)")
751    println("  - 平台线程:      传统 Thread / CachedThreadPool")
752    println()
753
754    try {
755        benchmarkSchedulingOverhead()
756        benchmarkMemoryOverhead()
757        benchmarkThreadPoolLimits()
758        benchmarkCpuIntensive()
759        benchmarkIOIntensive()
760        benchmarkMixed()
761
762        // 生成测试报告
763        generateReport()
764
765    } catch (e: Exception) {
766        println("测试异常: ${e.message}")
767        e.printStackTrace()
768    }
769}
770
771