ZIO栈应用通过REST请求操作Kafka Topic订阅/取消订阅失效问题

1. 取消订阅不生效的问题 #

• 你调用Consumer.unsubscribe只是修改了消费者的订阅状态，但是已经启动的消费流startStream是一个正在运行的ZSink执行逻辑，它不会因为消费者取消订阅就自动终止，流本身还在轮询拉取消息，所以还会继续消费。
• 你没有持有startStream.fork返回的Fiber句柄，调用stop的时候没办法中断正在运行的消费流，所以仅取消订阅没有实际作用。

2. 调用/start端点后读不到消息的问题 #

• startStream需要Console、Clock、Has[Consumer]三个依赖，你在HttpApp的执行上下文中没有提供完整的依赖层，特别是Console.live和Clock.live的环境没有正确注入，导致流虽然被fork，但是实际执行时缺少环境卡住，没办法真的拉取消息。
• 另外你每次调用/start都会fork一个新的消费流，但是同一个消费者实例不能同时启动多个消费流，会导致资源竞争，新的流没办法正常工作。

首先你需要在程序中维护一个全局的Fiber引用，用来持有当前运行的消费流的Fiber，方便stop的时候中断：

// 定义Ref存储当前消费流的Fiber，类型为Option[Fiber[Throwable, Unit]]
val runningConsumerFiber: ZIO[Any, Nothing, Ref[Option[Fiber[Throwable, Unit]]]] = Ref.make(None)

然后修改start和stop的接口逻辑：

private val app = HttpApp.fromEffectFunction{
  case Method.POST -> Root / "stop" => for {
    fiberRef <- ZIO.service[Ref[Option[Fiber[Throwable, Unit]]]]
    fiberOpt <- fiberRef.get
    // 先中断正在运行的消费流
    _ <- ZIO.foreach(fiberOpt)(_.interrupt)
    // 再取消订阅
    _ <- ZIO.serviceWith[Consumer](_.unsubscribe)
    // 清空Fiber引用
    _ <- fiberRef.set(None)
    _ <- zio.console.putStrLn("stopped")
  } yield Response.ok
  case Method.POST -> Root / "start" => for {
    fiberRef <- ZIO.service[Ref[Option[Fiber[Throwable, Unit]]]]
    // 判断有没有正在运行的流，避免重复启动
    existingFiber <- fiberRef.get
    _ <- if (existingFiber.isDefined) ZIO.unit else for {
      // 注入完整依赖启动流
      fiber <- startStream.provideSomeLayer(consumer ++ Console.live ++ Clock.live).fork
      _ <- fiberRef.set(Some(fiber))
      _ <- zio.console.putStrLn("started")
    } yield ()
  } yield Response.ok
}

最后修改main方法，把Ref也加入到依赖层中：

override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] = (for {
  fiberRef <- runningConsumerFiber
  _ <- server.make.use(_ => console.putStrLn("server started") *> ZIO.never)
    .provideCustomLayer(ServerChannelFactory.auto ++ EventLoopGroup.auto() ++ consumer ++ ZLayer.succeed(fiberRef))
} yield ()).exitCode

• zio-kafka的消费流是长期运行的ZIO效果，它的生命周期和启动它的Fiber绑定，不是和消费者的订阅状态绑定，停止消费必须中断对应的Fiber。
• ZIO的所有效果都需要完整的环境依赖才能正确执行，在HttpApp中运行的效果不会自动继承main方法的环境，需要显式注入需要的依赖。
• 同一个Kafka消费者实例不能同时运行多个消费流，重复启动会导致消费异常，需要做重复启动的判断。

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