使用zio-http, quill与postgres开发web服务

本文面向有一定scala和zio基础的读者。

zio是用Scala语言开发的一套框架，核心功能是并发管理和资源管理，近年来在scala社区中逐渐流行。zio-http是zio生态中的http库，原本是非官方项目，前段时间获得了zio官方支持。quill是zio生态中的数据库操作库，支持主流关系型数据库，当初转投zio社区还引发了不小的风波。

本文旨在提供一套基于zio生态的http服务快速开发方式。

访问数据库

引入依赖

    libraryDependencies ++= Seq(
      "dev.zio" %% "zio" % "2.0.15",
      "io.getquill" %% "quill-jdbc-zio" % "4.6.1",
      "org.postgresql" % "postgresql" % "42.5.4"
    )

向resources目录里添加application.conf文件，填写数据库连接配置。

db {
  dataSourceClassName = org.postgresql.ds.PGSimpleDataSource
  dataSource.user = postgres
  dataSource.portNumber = 5432
  dataSource.password = password
  connectionTimeout = 30000
}

创建Main.scala作为程序入口，引入ZIODefaultApp。

import zio.{Scope, ZIO, ZIOAppArgs, ZIOAppDefault}

object Main extends ZIOAppDefault {
  override def run: ZIO[ZIOAppArgs with Scope, Any, Any] = ???
}

使用quill读取上面的db配置，创建数据源DataSource。这里的ZLayer是zio提供的依赖管理工具，类型参数有三个，第一个代表了依赖的项，第二个代表了创建依赖过程中可能抛出的错误类型，第三个代表了最终创建出的项。这里第一个参数是Any代表不依赖其他任何项就可以创建DataSource。

val dsLayer: ZLayer[Any, Throwable, DataSource] =
  Quill.DataSource.fromPrefix("db")

接口创建quill context用于编译和运行时使用。通过ZLayer的类型参数可以很容易地看出它依赖了DataSource才能成功创建quill context。

val quillLayer: ZLayer[DataSource, Nothing, Quill.Postgres[CompositeNamingStrategy2[SnakeCase, PostgresEscape]]] =
      Quill.Postgres.fromNamingStrategy(CompositeNamingStrategy2(SnakeCase, PostgresEscape))

我们通过>>>操作符将上一个ZLayer的结果传递给下一个ZLayer，组合出新的ZLayer。可以看到dsLayer的第三个类型参数与quillLayer的第一类型参数相抵消，我们就得到了一个不需要任何依赖就能创建quill context的ZLayer。

val contextLayer: ZLayer[Any, Throwable, Quill.Postgres[CompositeNamingStrategy2[SnakeCase, PostgresEscape]]] =
  dsLayer >>> quillLayer

有了quill context就可以进行数据库访问。假设有这样一张表。

create table "user"
(
    user_id  serial primary key,
    username varchar(255) not null unique,
    password varchar(255) not null
);

我们可以在程序中创建一个同名case class映射到这张表上，并通过quill context引入相关数据库操作符。

final case class User(userId: Int, username: String, password: String)

final case class UserRepository(quill: Quill.Postgres[CompositeNamingStrategy2[SnakeCase, PostgresEscape]]) {

  import quill._

  def listUser: ZIO[Any, SQLException, Seq[User]] = run(query[User])
}

object UserRepository {
  val layer: ZLayer[Quill.Postgres[CompositeNamingStrategy2[SnakeCase, PostgresEscape]], Nothing, UserRepository] =
    ZLayer.fromFunction(UserRepository.apply _)
}

这里的query[User]会在编译器生成对应的SQL，尝试编译下即可在命令行里看到。

[info] /zio-http-quill-demo/src/main/scala/example/repository/UserRepository.scala:14:65: SELECT x."user_id" AS userId, x."username" AS username, x."password" AS password FROM "user" x
[info]   def listUser: ZIO[Any, SQLException, Seq[User]] = run(query[User])

到这里，数据库访问的相关工作已经基本就绪。

http服务

引入依赖

    libraryDependencies ++= Seq(
      "dev.zio" %% "zio-http" % "3.0.0-RC2",
      "dev.zio" %% "zio-json" % "0.6.0",
    )

zio-http底层使用netty，有较好的性能，编写时也很方便。下面是一个最简单的demo。

import zio._
import zio.http._

object Main extends ZIOAppDefault {

  val app: App[Any] = 
    Http.collect[Request] {
      case Method.GET -> Root / "text" => Response.text("Hello World!")
    }

  override val run =
    Server.serve(app).provide(Server.default)
}

这里Http.collect[Request]通过pattern match模式匹配设置路由和对应的处理逻辑。由于我们使用了zio，会换成zio-http提供的Http.collectZio[Request]。

import zio.json._

def httpApp(userRepository: UserRepository): Http[Any, Nothing, Request, Response] = {
  Http.collectZIO[Request] {
    case Method.GET -> Root / "user" / "list" =>
      val response = for {
        worlds <- userRepository
          .listUser
          .mapError(e => ErrorMsg("INTERNAL_ERROR", e.getMessage))
      } yield Response.json(worlds.toJson)
      response
        .catchAll(errorMsg => ZIO.succeed(
          Response.json(errorMsg.toJson)
        ))
  }
}

这里展通过访问数据库获取User列表，进行序列化后返回Response。自定义的ErrorMsg代表请求失败时的响应。toJson是zio-json库提供的功能，将case class序化列成json。

日志

引入依赖

    libraryDependencies ++= Seq(
      "dev.zio" %% "zio-logging" % "2.1.13",
      "dev.zio" %% "zio-logging-slf4j2" % "2.1.13",
      "org.slf4j" % "slf4j-api" % "2.0.7",
      "ch.qos.logback" % "logback-classic" % "1.4.8"
    )

在Main.scala中将ZIO.log的默认实现替换为zio-logging提供的组件。

import zio.logging.backend.SLF4J

object Main extends ZIOAppDefault {
  override val bootstrap: ZLayer[ZIOAppArgs, Any, Any] =
    Runtime.removeDefaultLoggers >>> SLF4J.slf4j
}

日志接口使用slf4j，实现使用logback。同时zio有自己的log操作符，我们添加zio-logging-slf4j2依赖将背后的实现替换为slf4j，这样我们在调用ZIO.log("xxx")时就会使用logback。

zio-http middleware

zio-http middleware是zio-http功能的扩展点，如果我们想要实现打印请求响应、链路追踪、超时和重试等功能，zio-http middleware就是一个非常良好的实现方式。

简单写一个在response里添加响应时间的middleware并通过@@操作符绑定到httpApp上。

import zio.http._
import zio.{Trace, ZIO}

class ResponseTimeMiddleware extends RequestHandlerMiddleware.Simple[Any, Nothing] {
  override def apply[R1 <: Any, Err1 >: Nothing](
      handler: Handler[R1, Err1, Request, Response]
  )(implicit trace: Trace): Handler[R1, Err1, Request, Response] =
    Handler.fromFunctionZIO[Request] { request =>
      for {
        startTime <- ZIO.succeed(System.currentTimeMillis())
        response <- handler.runZIO(request)
        endTime <- ZIO.succeed(System.currentTimeMillis())
      } yield response.addHeader(Header.Custom("X-Response-Time", s"${endTime - startTime}"))
    }
}

object Main extends ZIOAppDefault {
  override val run = {
    val responseTimeMiddleware = new ResponseTimeMiddleware()
    Server.serve(app @@ responseTimeMiddleware).provide(Server.default)
  }
}

实现一个简单的日志链路追踪

日志链路追踪是web后端服务常见的需求之一，我们已经了解了zio-http middleware和zio-logging的基础用法，现在结合二者的能力实现请求级别的日志链路追踪。

首先要复习下zio中的一个概念ZIOAspect，正如其名字aspect名字“切面”所言，可以包裹住一个ZIO调用并进行某种操作，常见的有日志、重试等。

import zio.http._
import zio.logging.LogAnnotation
import zio.{Trace, ZIO}

import java.util.UUID

class LoggingMiddleware extends RequestHandlerMiddleware.Simple[Any, Nothing] {
  override def apply[R1 <: Any, Err1 >: Nothing](
      handler: Handler[R1, Err1, Request, Response]
  )(implicit trace: Trace): Handler[R1, Err1, Request, Response] =
    Handler.fromFunctionZIO[Request] { request =>
      val h = for {
        _ <- ZIO.log(request.url.path.toString())
        response <- handler.runZIO(request)
      } yield response
      h @@ LogAnnotation.TraceId(UUID.randomUUID())
    }
}

这里调用LogAnnotation.TraceId()就是创建了一个ZIOAspect，具体作用是向这个Aspect中的日志上下文添加了traceId=XXX的信息，这样被包裹的ZIO操作中的ZIO.log就可以拿到相关信息并添加到日志中。

日志系统的底层是logback，zio-logging会向%kvp中添加我们传递的traceId上下文，我们将%kvp添加进logback.xml的pattern中。

<configuration>
  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
    <encoder>
      <pattern>[%level] [%kvp] - %msg %n</pattern>
    </encoder>
  </appender>
  <root level="INFO">
    <appender-ref ref="STDOUT"/>
  </root>
</configuration>

添加后，我们发起请求后就可以看到携带有traceId的日志，第一条日志是middleware打印的。并且我们在Http.collectZIO[Request]中调用ZIO.log也会打印traceId，因为已经被ZIOAspect包裹起来了，可以拿到上下文，第二条日志就是如此。

[INFO] [trace_id="2dbf5bd2-a856-4d21-945c-b42a08f3bdc0"] - /user/list
[INFO] [trace_id="2dbf5bd2-a856-4d21-945c-b42a08f3bdc0"] - return 2 users

需要注意，我们在middleware添加的ZIOAspect上下文是请求级别的，请求之间并不共享。

end

想要获取完整代码请访问 https://github.com/gcnyin/zio-http-quill-demo

访问数据库#

http服务#

日志#

zio-http middleware#

实现一个简单的日志链路追踪#

end#

访问数据库

http服务

日志

zio-http middleware

实现一个简单的日志链路追踪

end