如何异步中断fs2流？使用SignallingRef中断失败求助

你的代码里有几个关键问题导致流无法响应SignallingRef的变化，核心是错误地使用unsafeRunSync破坏了cats-effect的异步流程：

1. 错误的unsafeRunSync使用：

   • 在interruptWhen中，Stream.repeatEval(switch.map(_.get).unsafeRunSync)这行代码会在流构建阶段就执行unsafeRunSync，只会获取一次SignallingRef的初始值false，之后不会再监听它的变化，所以永远不会触发中断。
   • 同样，println(switch.map(_.get).unsafeRunSync.unsafeRunSync)也是错误的：switch是IO[SignallingRef[IO, Boolean]]，switch.map(_.get)得到的是IO[IO[Boolean]]，两次unsafeRunSync完全不符合纯函数IO范式，而且只会拿到初始值，不会随状态更新。

2. 没有正确引用SignallingRef实例：
   你的代码里switch是一个IO[SignallingRef]，但在流中没有正确获取并复用这个实例，反而反复用unsafeRunSync去获取，这会破坏异步上下文，导致无法监听状态变化。

下面是修正后的代码，完全遵循cats-effect和fs2的纯函数范式：

import cats.effect.IO
import fs2.Stream
import fs2.concurrent.SignallingRef
import scala.concurrent.ExecutionContext
import scala.concurrent.duration.DurationInt

implicit val contextShift = IO.contextShift(ExecutionContext.global)
implicit val timer = IO.timer(ExecutionContext.global)

// 先获取SignallingRef实例，用flatMap组合整个程序逻辑
val program: IO[Unit] = SignallingRef[IO, Boolean](false).flatMap { switch =>
  val streamingProgram: Stream[IO, Unit] = Stream
    .repeatEval(
      for {
        currentTime <- IO(java.time.LocalTime.now)
        switchValue <- switch.get
        _ <- IO(println(s"$currentTime $switchValue"))
      } yield ()
    )
    .metered(1.second)
    // 用SignallingRef的discrete流监听状态变化，值为true时触发中断
    .interruptWhen(switch.discrete)

  // 启动流并模拟延迟触发中断，你可以替换为自己的触发逻辑
  streamingProgram.compile.drain.start.flatMap { fiber =>
    IO.sleep(3.seconds) >> switch.set(true) >> fiber.join
  }
}

program.unsafeRunSync()

• 移除不当的unsafeRunSync：所有IO操作通过for-comprehension和flatMap组合，保持纯函数异步流程，确保能正确监听SignallingRef的状态变化。
• 复用SignallingRef实例：通过flatMap获取到switch实例后，直接在流中使用它的get方法获取当前状态，用discrete流监听状态变更。
• 正确使用interruptWhen：switch.discrete会返回一个流，每当SignallingRef的值变化时就发出新值，当值变为true时，interruptWhen会立即触发流中断。
• 模拟中断触发：代码中用IO.sleep(3.seconds) >> switch.set(true)模拟3秒后触发中断，你可以替换成控制台输入、外部事件等自定义触发逻辑。

运行这段代码后，你会看到类似输出：

16:10:15.678901 false
16:10:16.678901 false
16:10:17.678901 false

3秒后switch被设为true，流会立即终止，不再输出后续内容。

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