Cats-effect 3：并发fs2流运行一段时间后IO挂起求助

问题背景 #

• 运行环境：GCP e2-small（0.5 vCPU）实例
• 应用架构：fs2应用，仅启动io-compute-0和io-compute-1两个计算线程
• 核心业务流：两个并发执行的fs2.Stream
  • Stream 1：每3秒从Ref.of[F, SessionInfo]读取凭证，调用REST API处理数据，通过Sync[F].blocking将结果保存为zip文件到磁盘，同时将部分数据写入Ref.of[F, MyDTO]
  • Stream 2：每281秒从Ref.of[F, MyDTO]读取数据，通过Sync[F].blocking保存到另一磁盘文件
• 依赖配置：使用AsyncHttpClientCatsBackend，连接超时设为1分钟，但应用挂起数小时后未触发超时
• 故障现象：运行约1小时后完全挂起，线程dump显示两个io-compute线程均处于等待WorkStealingThreadPool（0x00000000e0ddac58）通知的状态

核心原因与解决方案 #

1. 计算线程池被阻塞IO耗尽 #

fs2默认的io-compute线程池是计算型线程池，设计用于处理非阻塞的计算/调度任务，不适合承载磁盘IO、同步网络调用等阻塞操作。在0.5vCPU的受限环境下，仅有的2个计算线程很容易被Sync[F].blocking标记的磁盘读写操作占满，导致线程池无可用线程处理定时调度、超时触发等核心逻辑，最终整个应用进入死锁状态。

修复措施：分离阻塞IO到专用线程池

为磁盘读写等阻塞操作创建独立的阻塞线程池，避免占用计算线程：

// cats-effect 3+ 示例：自定义阻塞线程池
import cats.effect.IO
import java.util.concurrent.Executors
import scala.concurrent.ExecutionContext

// 创建专用阻塞线程池（根据IO密集程度调整大小，建议4-8）
val blockingThreadPool = Executors.newFixedThreadPool(4)
val blockingEc = ExecutionContext.fromExecutor(blockingThreadPool)

// 替换原有的Sync[F].blocking，将阻塞操作绑定到专用线程池
def writeToFile(path: String, data: Array[Byte]): IO[Unit] =
  IO.blocking {
    java.nio.file.Files.write(java.nio.file.Paths.get(path), data)
  }.evalOn(blockingEc)

2. 超时未触发：超时逻辑未覆盖全链路 #

仅设置客户端连接超时不足以覆盖整个API请求流程（比如响应等待、数据处理阶段），且如果计算线程被阻塞，超时调度任务无法被执行，导致超时失效。

修复措施：为全API链路添加显式超时

为整个API调用流程添加顶层超时，确保即使某个环节阻塞，超时逻辑也能触发：

import scala.concurrent.duration._
import org.http4s._
import org.http4s.client._
import org.http4s.asynchttpclient.cats._

def fetchAndProcessData(session: SessionInfo): IO[Unit] =
  AsyncHttpClientCatsBackend[IO]().use { client =>
    client.expect[ApiResponse](Uri.unsafeFromString("https://your-api-endpoint.com"))
      .flatMap(processResponse) // 处理响应逻辑
      .flatMap(writeToZipFile) // 写入zip文件（用上述专用阻塞线程池）
  }.timeout(1.minute) // 为整个流程添加1分钟超时

3. 适配0.5vCPU环境的线程池参数 #

GCP e2-small的0.5vCPU属于共享CPU，默认的计算线程池大小（基于availableProcessors()）可能不适合这种受限环境，导致线程调度效率低下。

修复措施：调整JVM启动参数优化线程池

在应用启动时添加以下JVM参数，适配受限资源：

-Dcats.effect.io-compute.parallelism=1 \
-Dcats.effect.blocking.parallelism=4

• io-compute.parallelism：设置计算线程数为1，避免在0.5vCPU环境下过度抢占CPU资源
• blocking.parallelism：设置阻塞线程池大小为4，足够处理磁盘IO和网络阻塞操作

4. 线程dump的深层解读 #

从线程dump可见，两个io-compute线程均处于WAITING状态，说明：

• 计算线程池的任务队列已空，且没有新的任务被提交（比如Stream的定时触发任务因线程阻塞无法执行）
• 所有阻塞操作都占用了计算线程，导致调度器无法推进任何任务，最终应用挂起

5. Ref并发访问的优化 #

虽然Ref是线程安全的，但频繁的读写操作（每3秒一次）可能带来微小的同步开销，可通过以下方式优化：

• 如果MyDTO是大对象，改用Queue[F, MyDTO]传递数据，避免频繁的内存拷贝和Ref更新同步
• 确保Ref的更新逻辑是纯函数式的，无隐藏阻塞操作

内容的提问来源于stack exchange，提问作者tharindu_DG