如何使用Redisson与ZIO实现Redis列表事件监听

我们可以基于ZIO的异步回调能力和ZStream流处理能力完成封装，其中ZStream形式更适合持续事件的处理链路搭建，可以天然支持背压、算子转换、资源安全管理等能力。

依赖准备 #

首先确保项目引入如下核心依赖：

// build.sbt 配置示例
libraryDependencies ++= Seq(
  "dev.zio" %% "zio" % "2.0.21",
  "dev.zio" %% "zio-streams" % "2.0.21",
  "org.redisson" % "redisson" % "3.23.5"
)

封装为ZStream实现 #

我们通过ZIO的作用域（Scope）自动管理Redisson客户端和监听器的生命周期，避免资源泄漏：

import zio._
import zio.stream._
import org.redisson.Redisson
import org.redisson.api.{RQueue, ListAddListener}
import org.redisson.client.codec.StringCodec
import org.redisson.config.Config

object RedissonZioAdapter {
  /**
   * 监听Redisson队列的新增元素事件，返回ZStream
   * @param queueName 监听的队列名
   * @param redissonConfig Redisson配置，默认加载本地默认配置
   */
  def listenQueueEvents(queueName: String, redissonConfig: => Config = new Config()): ZStream[Any, Throwable, String] = {
    // 管理Redisson客户端生命周期：创建 + 退出时自动关闭
    val redissonResource = ZIO.acquireRelease(
      ZIO.attempt(Redisson.create(redissonConfig))
    )(client => ZIO.attempt(client.shutdown()).ignore)

    ZStream.scoped(redissonResource).flatMap { redisson =>
      ZStream.asyncScoped[Any, Throwable, String] { emit =>
        for {
          queue <- ZIO.attempt(redisson.getQueue[String](queueName, new StringCodec()))
          // 注册监听器，获取监听器ID用于后续释放
          listenerId <- ZIO.attempt(
            queue.addListener(new ListAddListener[String] {
              override def onListAdd(value: String): Unit = {
                // 将新增事件推送到流中
                emit(ZIO.succeed(Chunk(value)))
              }
            })
          )
          // 作用域结束时自动移除监听器
          _ <- ZIO.addFinalizer(ZIO.attempt(queue.removeListener(listenerId)).ignore)
        } yield ()
      }
    }
  }
}

完整使用示例 #

直接基于返回的ZStream搭建事件处理链路即可：

object MinioEventProcessor extends ZIOAppDefault {
  override def run: ZIO[Any, Throwable, Unit] = {
    RedissonZioAdapter.listenQueueEvents("minio_events")
      .filter(_.nonEmpty) // 可直接添加过滤、转换等流算子
      .tap(event => ZIO.logInfo(s"处理minio事件: $event")) // 替换为实际业务处理逻辑
      .runDrain // 持续运行流，直到程序退出
  }
}

注意事项 #

• 如果Redisson客户端已经在项目其他地方初始化，可直接传入已有的客户端实例，不需要重复创建
• 高吞吐场景下可以将ZStream.asyncScoped替换为ZStream.asyncScopedWithBackpressure，开启背压支持
• 如果仅需要单次监听事件，可使用ZIO.async封装为单个异步任务，持续事件场景优先选择ZStream

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