使用ZIO-Streams的Redis Stream消费者和生产者实现源码

21-03-23 banq

一个利用zio-streams和Redisson库来使用和产生事件到Redis流的示例应用程序,基于Java 11,使用ZIO-Streams的Redis Stream消费者和生产者实现。点击标题见Github。

 

ZIO/ZIO Streams库在Scala函数式编程世界中非常令人着迷,Monad风格。与Akka流相比,ZIO流更简单,更灵活且更具表现力。

以下是ZIO另外一个使用案例

要求:

  • 使用流逐行处理日志
  • 过滤掉不是ERROR或WARN的消息
  • 将流分成2个流:一个用于ERROR,另一个用于WARN消息
  • 根据一定的时间表处理错误消息,例如 每2秒
  • 批量处理WARN消息。例如,将10条消息分成一批

ZStream是ZIO Streams中的主要构造。首先,我们将从文件输入流创建ZStream:

ZStream.fromInputStream(fileInputStream)

您得到的是流的块[T]。您可以将Chunk [T]视为Array [T]的不变版本,但效率更高。

.chunks
.aggregate(ZSink.utf8DecodeChunk)
.aggregate(ZSink.splitLines)
.mapConcatChunk(identity)

然后,我们将Chunk [String]的流转换为String的流,并且每个元素都是日志文件中的一行消息。

ZIO Streams具有.tap函数,您可以使用该函数使流中的消息达到峰值。它是调试流应用的好工具。

.tap(data => putStrLn(s"> $data"))

然后我们过滤掉不是ERROR或WARN的消息:

.filter(isErrorWarning)

下一行代码基于isError谓词返回true或false ,将当前流分成2个流。数字4是缓冲区大小,因此两个子流可以以不同的速度运行直至缓冲区大小:

.partition(isError, 4)
val errorStream = leftStream
  .mapM(processError(_))
  .schedule(Schedule.fixed(2.second))

使用ZIO Schedule每2秒处理一次左流中的每个ERROR消息。

对于正确流中的WARN消息,我们将一次将它们分为6条消息并进行批量处理:

val warningStream = rightStream
  .aggregate(ZSink.collectAllN[String](10))
  .mapM(processWarning(_))

如果您更容易理解,您可以认为ZSink是消息生产者/使用者上下文中的消息使用者。(并认为ZStream是消息生成器)。ZSink通常用于ZIO Stream中的聚合功能。

最后,我们将2个流合并为一个并收集所有元素:

errorStream.merge(warningStream).runCollect

以下是完整代码:

import java.nio.file.{Files, Paths}

import zio._
import zio.console._
import zio.duration._
import zio.stream._

object LogStreamApp extends App {

  def isErrorWarning(data: String) = {
    data.contains("ERROR") || data.contains("WARN")
  }

  def isError(data: String): Boolean = {
    data.contains("ERROR")
  }

  def processError(data: String) = {
    putStrLn(s"process error message: ${data}") *>
      Task.succeed()
  }

  def processWarning(list: List[String]) = {
    putStrLn(s"process warning messages in batch: ${list.length} => $list") *>
      Task.succeed()
  }

  def run(args: List[String]): ZIO[ZEnv, Nothing, Int] = {
    val is = Files.newInputStream(Paths.get(ClassLoader.getSystemResource("prod_log.txt").toURI()))

    val theJob = (for {
      streams <- ZStream
        .fromInputStream(is)
        .chunks
        .aggregate(ZSink.utf8DecodeChunk)
        .aggregate(ZSink.splitLines)
        .mapConcatChunk(identity)
        .tap(data => putStrLn(s"> $data"))
        .filter(isErrorWarning)
        .partition(isError, 4)
    } yield streams).use {
      case (leftStream, rightStream) => {
        val errorStream = leftStream
          .mapM(processError(_))
          .schedule(Schedule.fixed(2.second))

        val warningStream = rightStream
          .aggregate(ZSink.collectAllN[String](10))
          .mapM(processWarning(_))

        errorStream.merge(warningStream).runCollect
      }
    }

    theJob.fold(_ => 1, _ => 0)
  }
}

the code at github.

猜你喜欢