码迷,mamicode.com
首页 > 其他好文 > 详细

Lagom 官方文档之随手记

时间:2020-07-26 19:04:18      阅读:102      评论:0      收藏:0      [点我收藏+]

标签:sem   地方   latest   discovery   root   代码   ber   catch   rod   

引言

Lagom是出品Akka的Lightbend公司推出的一个微服务框架,目前最新版本为1.6.2。Lagom一词出自瑞典语,意为“适量”。

?? https://www.lagomframework.com/documentation/1.6.x/scala/Home.html

Lagom框架坚持,微服务是按服务边界Boundary将系统切分为若干个组成部分的结果,这意味着要使它们与限界上下文Bounded Context、业务功能和模块隔离等要求保持一致,才能达到可伸缩性和弹性要求,从而易于部署和管理。因此,在设计微服务时应考虑大小是否“Lagom”,而非是否足够“Micro”。

Lagom框架大量使用了Jonas Bonér所著Reactive Microservices Architecture: Design Principles For Distributed Systems一书的设计理念和思想,所以推荐在使用Lagom之前先阅读此书。]

“船大好顶浪,船小好调头”——Jonas认为,将庞大的系统分割为若干独立的更小粒度的部分,同时将管理权限适当下放,可以使这些独立的部分更快地做出决断,以适应外部环境的不断变化。

?? Getting Start

Lagom开发环境要求:

  • JDK8
  • sbt 1.2.1以上版本(Lightbend推荐使用sbt,maven次之,故以下均使用sbt。)
  • 可用的互联网

Hello World

Lagom提供了Giter8模板,方便利用sbt构造一个Hello World项目结构,确保在开发前验证生成工具和项目已正确配置。Hello World包括Hello与Stream两个微服务,每个微服务包括API与实现两个子项目。Lagom会自动配置诸如持久化、服务定位等基础设施,并且支持发现并加载微服务的热更新。

sbt new lagom/lagom-scala.g8

hello                   → Project root
 └ hello-api            → hello api project
 └ hello-impl           → hello implementation project
 └ hello-stream-api     → hello-stream api project
 └ hello-stream-impl    → hello-stream implementation project
 └ project              → sbt configuration files
   └ build.properties   → Marker for sbt project
   └ plugins.sbt        → sbt plugins including the declaration for Lagom itself
 └ build.sbt            → Your project build file

使用sbt里的runAll一次性启动所有服务后(包括Cassandra、Kafka,若是手动则需要自己一个个启动包括基础服务在内的所有服务),在http://localhost:9000/api/hello/World处将得到显示了一行“Hello World”的页面。

  • 在Hello API里,HelloWorldService从Service派生,声明服务的API以及服务的回复消息。
  • 在API里,实现Service.descriptor方法,该方法返回一个Descriptor,用于定义服务的名称、REST端点路径、Kafka消息主题,以及REST路径与服务API方法的映射关系,等等。
  • 在Hello Impl里,HelloWorldServiceImpl从API里的HelloWorldService派生,定义服务的API。
  • 在服务API的定义里,将使用一个支持集群分片的Sharded、可持久化的Persistent、强类型Typed的Actor——HelloWorldBehavior采用Ask模式进行通信,按entityRef(id).ask[Message](replyTo => msg(replyTo)).map(reply => ...)的样式进行ask与map配对。其中,entityRef是该Actor在集群里的引用:clusterSharding.entityRefFor(HelloWorldState.typeKey, id)
  • HelloWorldBehavior实质就是一个EventSourcedBehavior,采用了State Pattern,定义了支撑服务的State、Command和Event以及相应的Handler。
  • 在Hello API内部,hello(id)实际是委托useGreeting(id)方法完成的,所以将World换作其他内容亦可。

服务与网络地址绑定

使用sbt将服务绑定到特定网络地址(默认是localhost):

lazy val biddingImpl = (project in file("biddingImpl"))
  .enablePlugins(LagomScala)
  .settings(lagomServiceAddress := "0.0.0.0")

端口号的分配机制

即便是将服务部署到不同的物理主机上,服务的端口号也将保持前后一致(总是使用特定的某个端口号),该端口号基于以下算法:

  • 先对项目Project的名称进行hash。
  • 将hash值投射到默认的端口范围[49152, 65535]内。
  • 如果没有其他项目申请同一个端口号,那么选定的该端口号将指定给该项目。如果发生冲突,那么将会按项目名称的字母顺序逐个递增地分配相邻的端口号。

通常情况下,无需关心上述细节,因为很少会发生这样的冲突。当然,也可以按下列方式手动指定端口号:

lazy val usersImpl = (project in file("usersImpl"))
  .enablePlugins(LagomScala)
  .settings(lagomServiceHttpPort := 11000)

如果嫌默认的端口范围[49152, 65535]不合适,可以指定端口范围,不过范围越窄,发生冲突的机率就越大:

lagomServicesPortRange in ThisBuild := PortRange(40000, 45000)

在开发模式下使用HTTPS

开发模式,是指在sbt或者maven支持下,启动和调试各种服务。在这种环境下,对代码做出修改后,Lagom将负责后续的编译和重新加载工作,相应的服务会自动重启。

在sbt里进行如下配置,将会使用一个自签名的证书为服务启用HTTPS并指定其端口,确保服务之间能通过HTTPS进行调用,但同时服务网关Service Gateway将仍旧只能使用HTTP:

lagomServiceEnableSsl in ThisBuild := true
lagomServiceHttpsPort := 20443

在客户端,则建议使用诸如Play-WS或者Akka-HTTP Client API这样的HTTPS Client框架。

服务定位子与网关

服务定位子

服务定位子Service Locator,是确保用于发现其他服务并与之联系的组件。Lagom内置的缺省Locator有以下特性:

  • 默认地址是localhost,可使用lagomServiceLocatorAddress in ThisBuild := "0.0.0.0"修改之。
  • 默认端口是9008,可使用lagomServiceLocatorPort in ThisBuild := 10000修改之。
  • 非Lagom的外部服务需要先注册:lagomUnmanagedServices in ThisBuild := Map("weather" -> "http://localhost:3333"),然后再使用。

网关

网关Gateway,相当于Service Locator的代理,用于防止对Locator的不当访问。Lagom内置的缺省Gateway有以下特性:

  • 默认地址是localhost,可使用lagomServiceGatewayAddress in ThisBuild := "0.0.0.0"修改之。
  • 默认端口是9000,可使用lagomServiceGatewayPort in ThisBuild := 9010修改之。
  • Lagom提供了一个基于Akka HTTP的网关(akka-http是默认的),一个旧式的基于Netty的网关,可使用lagomServiceGatewayImpl in ThisBuild := "netty"指定之。

启停与禁用

在开发环境下,使用runAll时默认会启动Locator与Gateway。需要手动启停时,分别执行lagomServiceLocatorStartlagomServiceLocatorStop任务即可。

如要需要禁用内置的Locater与Gateway,则使用lagomServiceLocatorEnabled in ThisBuild := false禁用之。之后便需要自己提供一个Locator的实现,并需要牢记每个服务的端口号以建立相互联系。

Cassandra服务

Cassandra是Apache提供的一个分布式、可扩展的NoSQL数据库。它以KeySpace为单位,其中包含若干个表Table或者列族Column Family,每个列族可以有不同的列(相当于RDBMS中的字段)并可自由添加列,每个行可以拥有不同的列,并支持索引。Cassandra支持的数据类型除常见的原生类型外,为List、Map和Set提供了直接支持,提供了TTL数据到期自动删除功能,并且可以自定义数据类型。

Lagom内置了一个Cassandra服务,作为Event Sourcing的事件持久化平台。

  • 默认端口是4000,是为避免与Cassandra默认端口9042冲突,可以用lagomCassandraPort in ThisBuild := 9042指定之。
  • 默认情况下生成的数据会在每次Cassandra服务运行期间被保存,可以用lagomCassandraCleanOnStart in ThisBuild := true使之在每次服务启动时清空数据库。
  • 默认使用文件dev-embedded-cassandra.yaml对Cassandra进行配置,可以使用lagomCassandraYamlFile in ThisBuild := Some((baseDirectory in ThisBuild).value / "project" / "cassandra.yaml")另行指定特定YAML配置文件。
  • Cassandra服务将运行在独立的JVM上,可以使用lagomCassandraJvmOptions in ThisBuild := Seq("-Xms256m", "-Xmx1024m", "-Dcassandra.jmx.local.port=4099")指定相应的JVM参数。
  • 默认的日志将输出至标准输出,默认级别为ERROR,暂时未提供配置措施。如果确需调整日志配置,则只有连接到一个本地的Cassandra实例再修改之。
  • 默认情况下Cassandra服务会最先启动,并预留20秒时间完成启动,可以使用lagomCassandraMaxBootWaitingTime in ThisBuild := 0.seconds修改之。
  • 使用lagomCassandraStartlagomCassandraStop手动启停。
  • 使用lagomCassandraEnabled in ThisBuild := false禁用之,再使用lagomUnmanagedServices in ThisBuild := Map("cas_native" -> "tcp://localhost:9042")注册本地Cassandra实例即可使用之。

Kafka服务

Kafka是Apache提供的一个分布式的流处理平台,简单讲可以理解为一个生产者-消费者结构的、集群条件下的消息队列Message Queue。每条消息流Stream以主题Topic作为唯一区别,每个Topic下可以有多个相同的分区Partition,分区里的消息都有一个Offset作为序号以确保按顺序被消费。Kafka依赖ZooKeeper提供的集群功能部署其节点。

Lagom内置了一个Kafka服务:

  • 默认端口为9092,可以用lagomKafkaPort in ThisBuild := 10000修改之
  • 依赖的ZooKeeper端口默认为2181,可以用lagomKafkaZookeeperPort in ThisBuild := 9999修改之
  • 默认使用文件kafka-server.properties配置Kafka运行参数,可以使用lagomKafkaPropertiesFile in ThisBuild := Some((baseDirectory in ThisBuild).value / "project" / "kafka-server.properties")另行指定配置文件。
  • Kafka服务将运行在独立的JVM上,可以使用lagomKafkaJvmOptions in ThisBuild := Seq("-Xms256m", "-Xmx1024m")指定相应的JVM参数。
  • 默认的日志将直接输出至文件,路径为<your-project-root>/target/lagom-dynamic-projects/lagom-internal-meta-project-kafka/target/log4j_output,而Kafka的提交日志将保存在<your-project-root>/target/lagom-dynamic-projects/lagom-internal-meta-project-kafka/target/logs
  • 使用lagomKafkaStartlagomKafkaStop手动启停。
  • 使用lagomKafkaEnabled in ThisBuild := false禁用之,再使用lagomKafkaAddress in ThisBuild := "localhost:10000"指定Kafka实例即可使用之。如果是本地实例,用lagomKafkaPort in ThisBuild := 10000指定端口即可。

?? 编写Lagom服务

服务描述子

每一个Lagom服务都由一个接口进行描述。该接口的内容不仅包括接口方法的声明和实现,同时还定义了接口的元数据如何被映射到底层的传输协议。

服务的每个接口方法都要求返回一个ServiceCall[Request, Response],其中Request或Response可以是Akka的NotUsed

trait ServiceCall[Request, Response] {
  def invoke(request: Request): Future[Response]
}

每一个Lagom服务在覆写的descriptor中都将返回一个服务描述子Service Descriptor。以下便是声明了一个叫作hello的服务,该服务提供了sayHello的API:

trait HelloService extends Service {
  def sayHello: ServiceCall[String, String]

  override def descriptor = {
    import Service._
    named("hello").withCalls(call(sayHello))
  }
}

Call标识符

每个服务调用都必须有一个唯一的标识符,以保证调用能最终映射到正确的API方法上。这个标识符可以是静态的一个字符串名称,也可以在运行时动态生成。默认情况下,API方法的名称即该调用的标识符。

强命名的标识符

使用namedCall指定强命名的标识符,服务hello里的API方法sayHello的调用名为hello,在REST架构下的相应路径为/hello

named("hello").withCalls(namedCall("hello", sayHello))

基于路径的标识符

使用pathCall指定基于路径的标识符,类似于字符串中用$引导的内插值,此处用:引导内插变量。Lagom为此提供了一个隐式的PathParamSerializer,用于从路径中提取StringIntBoolean或者UUID类型的内插变量。比如以下便是提取了路径中类型为long的orderId和类型为String的itemId值,作为参数传递给API方法。在作参数映射时,默认将按从路径中从左至右提取的顺序进行映射。

?? 这与ASP.NET MVC等一些HTTP框架采用的方法是类似的,所以要注意以构建RESTful应用的思路贯彻学习始终。

def getItem(orderId: Long, itemId: String): ServiceCall[NotUsed, Item]

override def descriptor = {
  import Service._
  named("orders").withCalls(pathCall("/order/:orderId/item/:itemId", getItem _))
}

提取查询串中的参数时,则使用的?起始、以&分隔的形式:

def getItems(orderId: Long, pageNo: Int, pageSize: Int): ServiceCall[NotUsed, Seq[Item]]

override def descriptor = {
  import Service._
  named("orders").withCalls(pathCall("/order/:orderId/items?pageNo&pageSize", getItems _))
}

在REST架构下,Lagom会努力正确实现上述映射,并且在有Request消息时使用POST方法,否则使用GET方法。

REST标识符

REST标识符用于完全REST形式的调用,和pathCall非常类似,区别只是REST标识符可指定HTTP调用方法:

def addItem(orderId: Long): ServiceCall[Item, NotUsed]
def getItem(orderId: Long, itemId: String): ServiceCall[NotUsed, Item]
def deleteItem(orderId: Long, itemId: String): ServiceCall[NotUsed, NotUsed]

def descriptor = {
  import Service._
  import com.lightbend.lagom.scaladsl.api.transport.Method
  named("orders").withCalls(
    restCall(Method.POST, "/order/:orderId/item", addItem _),
    restCall(Method.GET, "/order/:orderId/item/:itemId", getItem _),
    restCall(Method.DELETE, "/order/:orderId/item/:itemId", deleteItem _)
  )
}

消息

每个服务API都需要指定Request和Response的消息类型,可以用akka.NotUsed作为占位符,分为两种形式:

  • 严格消息Strict Message:这就是用Scala里的object或者case class表达的常见消息,它们将在内存的缓冲区中被序列化后进行传输。如果Request与Response均是这类严格消息,则调用将会是同步的,严格按请求-回复的顺序进行。
  • 流消息Streamed Message:流消息是一类特殊的消息,其类型为Akka-Stream里的Source。Lagom将使用WebSocket协议进行流的传输。如果Request与Response均是流消息时,任何一方关闭时,WebSocket将完全关闭。如果只有一方是流消息,则严格消息仍按序列化方式进行传输,而WebSocket将始终保持打开状态,直到另一个方向关闭为止。

以下分别是单向流和双向流消息的示例:

def tick(interval: Int): ServiceCall[String, Source[String, NotUsed]]

def descriptor = {
  import Service._
  named("clock").withCalls(pathCall("/tick/:interval", tick _))
}

def sayHello: ServiceCall[Source[String, NotUsed], Source[String, NotUsed]]

def descriptor = {
  import Service._
  named("hello").withCalls(call(this.sayHello))
}

消息的序列化

Lagom通过定义隐式的MessageSerializer,为call、namedCall、pathCall和restCall提供了消息的序列化支持。对String类型的消息和Play框架JSON格式的消息,Lagom提供了内置的序列化器。除此以外,可以自定义序列化器。(参考:消息序列化器

使用Play-JSON时,通常是用case class和companion object配合,case class定义消息的结构,companion object定义消息的格式:

case class User(
    id: Long,
    name: String,
    email: Option[String]
)

object User {
  import play.api.libs.json._
  implicit val format: Format[User] = Json.format[User]
}

对应的JSON结果为:

{
  "id": 12345,
  "name": "John Smith",
  "email": "john.smith@example.org"
}

属性可以是Option,这样如果为None,则Play-JSON在序列化时不会解析它、反序列化时不会生成该属性。如果case class还内嵌了其他case class,则被嵌入的case class也需要定义它的format。

实现服务

服务的实现,即实现之前声明的服务描述子trait。对服务API中声明的每个方法,使用ServiceCall的工厂方法apply,传入一个Request => Future[Response],返回一个ServiceCall

(?? 注意:Lagom大量使用函数作为返回值,从而充分发挥了FP组合高阶函数的优势。)

class HelloServiceImpl extends HelloService {
  override def sayHello = ServiceCall { name => Future.successful(s"Hello $name!") }
}

使用流消息

当消息不是普通的严格消息而是流消息时,需要使用Akka Stream来处理它。对应前面单向流的例子,它的实现如下:

override def tick(intervalMs: Int) = ServiceCall { tickMessage =>
  Future.successful(
    Source
      .tick(
        // 消息被发送前的延迟
        intervalMs.milliseconds,
        // 消息发送的间隔
        intervalMs.milliseconds,
        // 将被发送的消息
        tickMessage
      )
      .mapMaterializedValue(_ => NotUsed)
  )
}

处理消息的头部信息

如果某些时候需要处理消息的头部信息(通常是HTTP),那么Lagom提供了ServiceCall的派生类ServerServiceCall作为支持。ServerServiceCall将ServiceCall里的Header单独取出,提供了invokeWithHeaders方法,该方法第一个参数是RequestHeader,另一个参数才是Request本身,这样直接将invoke委托给invokeWithHeaders,从而方便在函数体中使用handleRequestHeaderhandleResponseHeader对头部信息进行处理(尽管ServiceCall本身也支持这2个处理函数)。

override def sayHello = ServerServiceCall { (requestHeader, name) =>
  val user = requestHeader.principal
    .map(_.getName)
    .getOrElse("No one")
  val response = s"$user wants to say hello to $name"

  val responseHeader = ResponseHeader.Ok.withHeader("Server", "Hello service")

  Future.successful((responseHeader, response))
}

ServerServiceCall工厂方法有一个版本可以同时处理Request与Response的头部信息,但也有不带处理头部信息的版本。后者虽然看起来与ServiceCall没什么区别,从而显得多此一举,但实际这是为满足组合服务调用的需求而存在的。

组合服务调用

组合服务调用,类似于把ServerServiceCall通过依赖注入传递给需要包裹在外层的日志、权限、过滤等切面服务,从而实现AOP切入到核心的服务API调用上。

在AOP切入的实现上,Lagom采取了组合高阶函数的方法。这就象是抹了一层又一层奶油的生日蛋糕,只有切开了才能看到最里层真正想要吃到的蛋糕。相比使用共享的线程变量等方法,不仅通过类型系统发挥了编译检查的优势而更加安全,并且还通过构造表达式树而提供了延迟计算的功能。

// AOP: Log
def logged[Request, Response](serviceCall: ServerServiceCall[Request, Response]) =
  ServerServiceCall.compose { requestHeader =>
    println(s"Received ${requestHeader.method} ${requestHeader.uri}")
    serviceCall
  }

override def sayHello = logged(ServerServiceCall { name =>
    Future.successful(s"Hello $name!")
  })

// AOP: Authentication
trait UserStorage {
  def lookupUser(username: String): Future[Option[User]]
}

def authenticated[Request, Response](serviceCall: User => ServerServiceCall[Request, Response]) = {
  // composeAsync允许异步地返回要调用的服务API
  ServerServiceCall.composeAsync { requestHeader =>
    // First lookup user
    val userLookup = requestHeader.principal
      .map(principal => userStorage.lookupUser(principal.getName))
      .getOrElse(Future.successful(None))

    // Then, if it exists, apply it to the service call
    userLookup.map {
      case Some(user) => serviceCall(user)
      case None       => throw Forbidden("User must be authenticated to access this service call")
    }
  }
}

override def sayHello = authenticated { user =>
  ServerServiceCall { name =>
    // 注意:因为闭包,此处可访问经AOP切入带来的user
    Future.successful(s"$user is saying hello to $name")
  }
}

依赖注入

依赖注入,Dependency Injection,是把本应由服务承担的创建其自身依赖的责任移交给外部的框架,改由DI框架负责生产对象并维护彼此的关联,最终形成完整的对象图(Object Graph),从而达到公示服务所需依赖、消灭代码中硬编码的new操作、降低耦合度的目的。

Cake Pattern

在Scala语言里最常见的一个DI模式,是用trait的Self Type特性实现的蛋糕模式(Thin Cake Pattern)。

更复杂的示例,请参见 ?? Real-World Scala: Dependency Injection

trait Stuffing {
  val stuffing: String
}

// 使用Self Type特性,限定继承了Cake的子类,必须同时也继承了Stuffing
// 此处Stuffing还可以with其他trait,从而实现多重注入
trait Cake { this: Stuffing =>
  def flavour: String = this.stuffing
}

object LemonCake extends Cake with Stuffing { ... }

Reader Monad

在FP的世界,实现DI的另一个选择是Reader Monad。

case class Reader[R, A](run: R => A) {
  def map[B](f: A => B): Reader[R, B] =
    Reader(r => f(run(r)))

  def flatMap[B](f: A => Reader[R, B]): Reader[R, B] =
    Reader(r => f(run(r)).run(r))
}

def balance(accountNo: String) = Reader((repo: AccountRepository) => repo.balance(accountNo))

DI框架MacWire基本用法

?? Dependency Injection in Scala using MacWire
?? MacWire使用指南中涉及的参考知识:轨道交通编组站

Lagom使用了MacWire作为默认的DI框架(当然也可以换Spring之类的其他DI框架)。

MacWire主要使用wire[T]进行DI,它会从标注了@Inject的构造子、非私有的主要构造子、Companion Object里的apply()方法里查找依赖关系。然后再根据依赖项的类型,按以下顺序查找匹配的参数项:

  • 先在当前块、闭合函数或匿名函数的参数中寻找唯一值;
  • 然后在闭合类型的参数、属性里寻找唯一值;
  • 最后在父类型里寻找唯一值。

使用MacWire相关事项:

  • 如果是隐式参数,MacWire将会跳过它,使它仍按Scala语法关于隐式参数的方法进行处理。

  • 只要类型匹配,此处的值可以是vallazy val或者用def定义的无参函数。

  • 需要根据条件使用不同的依赖项实现时,用类似lazy val component = if (condition) then wire[implementationA] else wire[implementationB]的方法,定义好component的值即可。

  • 需要在依赖项上切入拦截器Interceptor时,先在依赖项定义处声明一个拦截器def logInterceptor : Interceptor,再用lazy val component = logInterceptor(wire[Component])绑定拦截声明,最后在使用依赖项进行实现的地方用lazy val logInterceptor = { ... }给出具体实现即可。

    trait ShuntingModule {
      lazy val pointSwitcher: PointSwitcher =
            logEvents(wire[PointSwitcher])
      lazy val trainCarCoupler: TrainCarCoupler =
            logEvents(wire[TrainCarCoupler])
      lazy val trainShunter = wire[TrainShunter]
    
      def logEvents: Interceptor
    }
    
    object TrainStation extends App {
      val modules = new ShuntingModule
          with LoadingModule
          with StationModule {
    
          lazy val logEvents = ProxyingInterceptor { ctx =>
            println("Calling method: " + ctx.method.getName())
            ctx.proceed()
          }
      }
    
      modules.trainStation.prepareAndDispatchNextTrain()
    }
    
  • 默认情况下,使用lazy val component = wire[Component]声明在当前范围内唯一的依赖项(Singleton)。如果需要在每次调用时都创建新的依赖项(Dependent),则换作def即可。

  • 除Singleton和Dependent两种生命周期外,MacWire还支持其他形式的生命周期,方法类似使用拦截器。

    trait StationModule extends ShuntingModule with LoadingModule {
      lazy val trainDispatch: TrainDispatch =
            session(wire[TrainDispatch])
      lazy val trainStation: TrainStation =
            wire[TrainStation]
    
      def session: Scope
    }
    
    object TrainStation extends App {
      val modules = new ShuntingModule
          with LoadingModule
          with StationModule {
    
          lazy val session = new ThreadLocalScope
      }
    
      // implement a filter which attaches the session to the scope
      // use the filter in the server
    
      modules.trainStation.prepareAndDispatchNextTrain()
    }
    
  • 对需要参数进行构造的依赖项,使用lazy val component = (parameters) => wire[Component]这样的工厂方法进行声明,展开后将变成new Component(parameters)。对在trait里声明的依赖项,则改用def component(parameters: Parameter) = wire[Component]的方式定义工厂方法。

  • 在具体使用参数化的依赖项时,使用wireWith(real_parameter)代入实际参数。

  • 需要区别同一类依赖项的不同实例时,可以用trait作为标记tag。一种方法是用new Component(...) with tag细化其子类型,另一种是在声明依赖值的类型后面用@@ tag作标记。

    trait Regular
    trait Liquid
    
    class TrainStation(
      trainShunter: TrainShunter,
      regularTrainLoader: TrainLoader with Regular,  
      liquidTrainLoader: TrainLoader with Liquid,
      trainDispatch: TrainDispatch) { ... }
    
    lazy val regularTrainLoader = new TrainLoader(...) with Regular
    lazy val liquidTrainLoader = new TrainLoader(...) with Liquid
    lazy val trainStation = wire[TrainStation]
    
    ///////////////////////////////////////
    
    class TrainStation(
      trainShunter: TrainShunter,
      regularTrainLoader: TrainLoader @@ Regular,  
      liquidTrainLoader: TrainLoader @@ Liquid,
      trainDispatch: TrainDispatch) { ... }
    
    lazy val regularTrainLoader = wire[TrainLoader].taggedWith[Regular]
    lazy val liquidTrainLoader = wire[TrainLoader].taggedWith[Liquid]
    lazy val trainStation = wire[TrainStation]
    

组装完整的应用程序

在Lagom中使用MacWire进行DI是可行的,但前述的服务定位子基本上只能满足开发环境的需求,生产环境还是要换用Akka Discovery Service Locator之类更强大的定位子框架。

在完成组件的拼装后,还需要一个启动子启动应用程序。Lagom为此提供了Play框架ApplicationLoader的简化版本LagomApplicationLoader。其中的loadDevModeload是必须的,前者用with LagomDevModeComponents加载Lagom内嵌的服务定位子,用于开发环境;后者可以指定生产环境下的服务定位子,如果返回NoServiceLoader将意味着不提供服务定位。而describeService是可选的,它主要用于声明服务的API,为外部管理框架或者脚本语言提供服务的Meta信息,以方便进行动态配置。服务无数据Metadata,又称为ServiceInfo,主要包括服务的名称以及一个访问控制列表ACL。通常这些Metadata会由框架自动生成,前提是在使用withCalls声明描述子时,用withAutoAcl(true)激活即可。

import com.lightbend.lagom.scaladsl.server._
import com.lightbend.lagom.scaladsl.api.ServiceLocator
import com.lightbend.lagom.scaladsl.devmode.LagomDevModeComponents

class HelloApplicationLoader extends LagomApplicationLoader {
  override def loadDevMode(context: LagomApplicationContext) =
    new HelloApplication(context) with LagomDevModeComponents

  override def load(context: LagomApplicationContext) =
    new HelloApplication(context) {
      override def serviceLocator = ServiceLocator.NoServiceLocator
    }

  override def describeService = Some(readDescriptor[HelloService])
}

最后,只需在配置application.conf里用play.application.loader = com.example.HelloApplicationLoader指定启动子,即完成了应用程序的装配。

Lagom预置的组件类型

Lagom根据用途不同,通常将组件分为若干类型:

  • Service Components:定义服务时所需要的基本模板
  • Persistence and Cluster Components:与Akka Cluster、Akka Persistence协作,实现Cluster、ES和CQRS所需要的组件模板
  • Broker API Components:与Kafka协作,进行消息的生产和消费所需要的组件模板
  • Service Locator Components:实现服务定位子的组件模板
  • Third party Components:与诸如Web Service Client之类进行协作的组件模板

?? 详细列表:https://www.lagomframework.com/documentation/1.6.x/scala/ScalaComponents.html

消费服务

基本步骤

定义并实现服务后,该服务即可被其他服务或其他类型的客户消费使用。Lagom将根据服务描述子的内容,通过使用ServiceClientimplement宏进行绑定,自动生成一个调用该服务的框架,然后就可以象调用本地对象的方法一样invoke服务提供的API了。

// 先绑定HelloService
abstract class MyApplication(context: LagomApplicationContext)
    extends LagomApplication(context)
    with AhcWSComponents {
  lazy val helloService = serviceClient.implement[HelloService]
}

// 然后在另一个服务MyService里消费HelloService
class MyServiceImpl(helloService: HelloService)(implicit ec: ExecutionContext) extends MyService {
  override def sayHelloLagom = ServiceCall { _ =>
    val result: Future[String] = helloService.sayHello.invoke("Lagom")

    result.map { response => s"Hello service said: $response" }
  }
}

消费流服务

当服务使用了流消息时,Lagom将在消费端使用带有最大帧长度参数的WebSocket进行通信。该参数定义了可以发送的消息的最大尺寸,可以在application.conf中配置。

#This configures the websocket clients used by this service.
#This is a global configuration and it is currently not possible to provide different configurations if multiple websocket services are consumed.
lagom.client.websocket {
  #This parameter limits the allowed maximum size for the messages flowing through the WebSocket. A similar limit exists on the server side, see:
  #https://www.playframework.com/documentation/2.6.x/ScalaWebSockets#Configuring-WebSocket-Frame-Length
  frame.maxLength = 65536
}

断路器

断路器Circuit Breaker,如同电路保险丝,可以在服务崩溃时迅速熔断,从而避免产生更大面积的连锁反应。

断路器有以下3种状态:

技术图片

  • 闭合状态 Closed:通常情况下,断路器处于闭合状态,保证服务调用正常通过。
    • 由于触发异常或者调用超过设定的call-timeout,导致失败计数值增长,在超过设定的max-failures,断路器跳到断开状态。
    • 由于半开状态下试探成功,失败计数值被重置归零。
  • 断开状态 Open:这种情况下,断路器处于断开状态,服务将始终不可用。
    • 所有的服务调用,都会得到一个CircuitBreakerOpenException异常而快速失败。
    • 在超过设定的reset-timeout后,断路器跳入半开状态。
  • 半开状态 Half-Open:这种情况下,断路器将尝试恢复到闭合状态。
    • 除第一个调用将被允许通过,以试探服务是否恢复可用外,后续调用将继续被快速失败所拒绝。
    • 如果试探成功,断路器将通过重置跳到闭合状态,恢复服务的可用性;如果试探失败,断路器将回到断开状态,然后等待下一个reset-timeout周期后再重新进行试探。

Lagom为所有消费端对服务的调用都默认启用了断路器。尽管断路器在消费端配置并使用,但用于绑定到服务的标识符则要由服务提供者定义和确定相应粒度。默认情况下,一个断路器实例将覆盖对一个服务所有API的调用。但通过设置断路器标识符,可以为每个API方法设置唯一的断路器标识符,以便为每个API方法使用单独的断路器实例。或者通过在某几个API方法上设置使用相同的标识符,实现对API调用的断路保护分组。

下例中,sayHi将使用默认的断路器,而hiAgain将使用断路器hello2

def descriptor: Descriptor = {
  import Service._

  named("hello").withCalls(
    namedCall("hi", this.sayHi),
    namedCall("hiAgain", this.hiAgain).withCircuitBreaker(CircuitBreaker.identifiedBy("hello2"))
  )
}

对应的application.conf中配置如下:

lagom.circuit-breaker {
  # will be used by sayHi method
  hello.max-failures = 5

  # will be used by hiAgain method
  hello2 {
    max-failures = 7
    reset-timeout = 30s
  }

  # Change the default call-timeout will be used for both sayHi and hiAgain methods
  default.call-timeout = 5s
}

默认情况下,Lagom的客户端会将所有4字头和5字头的HTTP响应都映射为相应的异常,而断路器会将所有的异常都视作失败,从而触发熔断。通过设置白名单,可以忽略某些类型的异常。断路器完整的断路器配置选项如下:

# Circuit breakers for calls to other services are configured
# in this section. A child configuration section with the same
# name as the circuit breaker identifier will be used, with fallback
# to the `lagom.circuit-breaker.default` section.
lagom.circuit-breaker {
  # Default configuration that is used if a configuration section
  # with the circuit breaker identifier is not defined.
  default {
    # Possibility to disable a given circuit breaker.
    enabled = on

    # Number of failures before opening the circuit.
    max-failures = 10

    # Duration of time after which to consider a call a failure.
    call-timeout = 10s

    # Duration of time in open state after which to attempt to close
    # the circuit, by first entering the half-open state.
    reset-timeout = 15s

    # A whitelist of fqcn of Exceptions that the CircuitBreaker
    # should not consider failures. By default all exceptions are
    # considered failures.
    exception-whitelist = []
  }
}

测试服务

Lightbend推荐使用ScalaTest和Spec2作为Lagom的测试框架。类似Akka ActorTestKit,可以使用Lagom提供的ServiceTest工具包对服务进行测试。

测试单个服务

为每个测试创建一个单独的服务实例,其关键步骤包括:

  • 使用Scala的异步测试支持AsyncWordSpec。
  • 使用ServiceTest.withServer(config)(lagomApplication)(testBlock)的结构进行测试。
  • 为上一步中的lagomApplication混入一个LocalServiceLocator,以启用默认的定位子服务。
  • 在testBlock中使用一个客户端进行服务调用。
import com.lightbend.lagom.scaladsl.server.LocalServiceLocator
import com.lightbend.lagom.scaladsl.testkit.ServiceTest
import org.scalatest.AsyncWordSpec
import org.scalatest.Matchers

// 1. 启用AsyncWordSpec
class HelloServiceSpec extends AsyncWordSpec with Matchers {
  "The HelloService" should {
    // 2. 使用ServiceTest.withServer进行测试
    "say hello" in ServiceTest.withServer(ServiceTest.defaultSetup) { ctx =>
      // 3. 启用默认的服务定位子
      new HelloApplication(ctx) with LocalServiceLocator
    } { server =>
      // 4. 创建一个客户端进行服务调用
      val client = server.serviceClient.implement[HelloService]

      client.sayHello.invoke("Alice").map { response =>
        response should ===("Hello Alice!")
      }
    }
  }
}

若要由多个测试共享一个服务实例,则需要使用ServiceTest.startServer()代替withServer(),然后在beforeAll与afterAll中启动和停止服务。

import com.lightbend.lagom.scaladsl.server.LocalServiceLocator
import com.lightbend.lagom.scaladsl.testkit.ServiceTest
import org.scalatest.AsyncWordSpec
import org.scalatest.Matchers
import org.scalatest.BeforeAndAfterAll

class HelloServiceSpec extends AsyncWordSpec with Matchers with BeforeAndAfterAll {
  lazy val server = ServiceTest.startServer(ServiceTest.defaultSetup) { ctx =>
    new HelloApplication(ctx) with LocalServiceLocator
  }
  lazy val client = server.serviceClient.implement[HelloService]

  "The HelloService" should {
    "say hello" in {
      client.sayHello.invoke("Alice").map { response =>
        response should ===("Hello Alice!")
      }
    }
  }

  protected override def beforeAll() = server

  protected override def afterAll() = server.stop()
}

若要启用Cluster、PubSub或者Persistence支持,则需要在withServer第1个参数中启用。若需要调用其他服务,可以在第2个参数中构造要调用服务的Stub或者Mock。注意事项包括:

  • 持久化功能只能在Cassandra与JDBC之间二者择其一。
  • 使用withCassandra或者withJdbc启用Persistence后。会自动启用Cluster
  • 无论自动或手动启用Cluster,都会启动PubSub。而PubSub不能手动启用。
lazy val server = ServiceTest.startServer(ServiceTest.defaultSetup.withCluster) { ctx =>
  new HelloApplication(ctx) with LocalServiceLocator {
    override lazy val greetingService = new GreetingService {
      override def greeting = ServiceCall { _ =>
        Future.successful("Hello")
      }
    }
  }
}

在测试时使用TLS

Lagom没有为HTTPS提供客户端框架,因此只有借用Play-WS、Akka HTTP或者Akka gRPC等框架创建使用SSL连接的客户端。而在服务端,可以使用withSsl激活SSL支持,随后框架将会为测试端自动打开一个随机的端口并提供一个javax.net.ssl.SSLContext类型的上下文环境。接下来,客户端就可以使用testServer提供的httpsPortsslContext连接到服务端并发送Request。Lagom测试工具提供的证书仅限于CN=localhost,所以该上下文SSLContext也只会信任本地的testServer,这就要求在发送请求时也设置好相应的权限,否则服务器将会拒绝该请求。目前,Lagom还无法为测试服务器设置不同的SSL证书。

"complete a WS call over HTTPS" in {
  val setup = defaultSetup.withSsl()
  ServiceTest.withServer(setup)(new TestTlsApplication(_)) { server =>
    implicit val actorSystem = server.application.actorSystem
    implicit val ctx         = server.application.executionContext
    // To explicitly use HTTPS on a test you must create a client of your own
    // and make sure it uses the provided SSLContext
    val wsClient = buildCustomWS(server.clientSslContext.get)
    // use `localhost` as authority
    val url = s"https://localhost:${server.playServer.httpsPort.get}/api/sample"
    val response = wsClient.url(url).get().map { _.body[String] }
    whenReady(response, timeout) { r => r should be("sample response") }
  }
}

测试流消息

在测试支持流消息的服务时,需要搭配Akka Streams TestKit进行测试。

"The EchoService" should {
  "echo" in {
    // Use a source that never terminates (concat Source.maybe)
    // so we don‘t close the upstream, which would close the downstream
    val input = Source(List("msg1", "msg2", "msg3")).concat(Source.maybe)
    client.echo.invoke(input).map { output =>
      val probe = output.runWith(TestSink.probe(server.actorSystem))
      probe.request(10)
      probe.expectNext("msg1")
      probe.expectNext("msg2")
      probe.expectNext("msg3")
      probe.cancel
      succeed
    }
  }
}

测试持久化实体

在服务测试中,可以通过额外编写PersistEntityTestDriver,使用持久化实体Persistent Entity进行与数据库无关的功能测试。

消息序列化器

Play、Akka和Lagom均出自Lightbend,因此Lagom使用Play JSON作为消息的序列化框架,算是开箱即用了。

选配序列化器

Lagom的序列化器,通常是与服务描述子放在一起的一个MessageSerializer类型的隐式变量。也可以在withCalls()声明服务API时,在call、namedCall、pathCall、restCall或者topic方法里显式地指定分别用于Request和Response的序列化器。

Lagom通过借用Play的JSON序列化器,对case class进行JSON格式的序列化。该JSON序列化器有一个主要方法是jsValueFormatMessageSerializer,可以使用它在case classscompanion object里指定其他格式的JSON模板。同时,Lagom的MessageSerializer也为NotUsedDoneString等类型提供了缺省的非JSON格式的序列化支持,比如MessageSerializer.StringMessageSerializer

trait HelloService extends Service {
  def sayHello: ServiceCall[String, String]

  override def descriptor = {
    import Service._

    named("hello").withCalls(
      call(sayHello)(
        MessageSerializer.StringMessageSerializer,
        MessageSerializer.StringMessageSerializer
      )
    )
  }
}

在companion object里定义不同的JSON格式,随后在服务描述子中显式地选择使用:

import play.api.libs.json._
import play.api.libs.functional.syntax._

case class MyMessage(id: String)

object MyMessage {
  implicit val format: Format[MyMessage] = Json.format
  val alternateFormat: Format[MyMessage] = {
    // 将id映射为JSON串里的identifier
    (__ \ "identifier")
      .format[String]
      .inmap(MyMessage.apply, _.id)
  }
}

trait MyService extends Service {
  def getMessage: ServiceCall[NotUsed, MyMessage]
  def getMessageAlternate: ServiceCall[NotUsed, MyMessage]

  override def descriptor = {
    import Service._

    named("my-service").withCalls(
      call(getMessage),
      call(getMessageAlternate)(
        // Request的序列化器
        implicitly[MessageSerializer[NotUsed, ByteString]],
        // Response的序列化器
        MessageSerializer.jsValueFormatMessageSerializer(
          implicitly[MessageSerializer[JsValue, ByteString]],
          // 指定JSON格式
          MyMessage.alternateFormat
        )
      )
    )
  }
}

自定义序列化器

Lagom的trait MessageSerializer也可以用来实现自定义的序列化器,派生的StrictMessageSerializer和StreamedMessageSerializer分别用于严格消息与流消息。其中,严格消息的序列化类型为二进制串ByteString,而流消息的则是Source[ByteString, _]

在实现自定义的序列化器之前,有几个关键性概念:

  • 消息协议 Message Protocols:包括内容类型,字符集和版本等3个可选属性,这3个属性将被投射到HTTP头部信息中的Content-TypeAccept,以及MIME Type Scheme的版本号,或者根据服务的配置方式直接从URL中提取。
  • 内容协商 Content Negotiation:Lagom镜像了HTTP的内容协商能力,保证服务器一方能根据客户端发出请求所使用的消息协议,采用对应协议进行通信。
  • 协商序列化器 Negotiation Serializer:使用内容协商后,MessageSerializer将不再直接承担序列化与反序列化职责,改由NegotiatedSerializerNegotiatedDeserializer负责。

内容协商在多数情况下并不是必要的,所以并不是一定需要实现的。

第一步:定义不同协议对应的序列化与反序列化器
/* ------------ String Protocol ------------ */
import akka.util.ByteString
import com.lightbend.lagom.scaladsl.api.deser.MessageSerializer.NegotiatedSerializer
import com.lightbend.lagom.scaladsl.api.transport.DeserializationException
import com.lightbend.lagom.scaladsl.api.transport.MessageProtocol
import com.lightbend.lagom.scaladsl.api.transport.NotAcceptable
import com.lightbend.lagom.scaladsl.api.transport.UnsupportedMediaType

// 注意:`charset`由构造子传入,传递给MessageProtocol用于定义协议使用的字符集。
class PlainTextSerializer(val charset: String) extends NegotiatedSerializer[String, ByteString] {
  override val protocol = MessageProtocol(Some("text/plain"), Some(charset))

  def serialize(s: String) = ByteString.fromString(s, charset)
}

import com.lightbend.lagom.scaladsl.api.deser.MessageSerializer.NegotiatedDeserializer

class PlainTextDeserializer(val charset: String) extends NegotiatedDeserializer[String, ByteString] {
  def deserialize(bytes: ByteString) =
    bytes.decodeString(charset)
}

/* ------------ JSON Protocol ------------ */
import play.api.libs.json.Json
import play.api.libs.json.JsString

class JsonTextSerializer extends NegotiatedSerializer[String, ByteString] {
  override val protocol = MessageProtocol(Some("application/json"))

  def serialize(s: String) =
    ByteString.fromString(Json.stringify(JsString(s)))
}

import scala.util.control.NonFatal

class JsonTextDeserializer extends NegotiatedDeserializer[String, ByteString] {
  def deserialize(bytes: ByteString) = {
    try {
      Json.parse(bytes.iterator.asInputStream).as[String]
    } catch {
      case NonFatal(e) => throw DeserializationException(e)
    }
  }
}
第二步:集成不同协议的序列化器
import com.lightbend.lagom.scaladsl.api.deser.StrictMessageSerializer
import scala.collection.immutable

class TextMessageSerializer extends StrictMessageSerializer[String] {
  // 支持的协议
  override def acceptResponseProtocols = List(
    MessageProtocol(Some("text/plain")),
    MessageProtocol(Some("application/json"))
  )

  /* ------ Serializer -----*/
  // Client发出Request时还不需要协商协议,故使用最简单的文本协议
  def serializerForRequest = new PlainTextSerializer("utf-8")

  def serializerForResponse(accepted: immutable.Seq[MessageProtocol]) = accepted match {
    case Nil => new PlainTextSerializer("utf-8")
    case protocols =>
      protocols
        .collectFirst {
          case MessageProtocol(Some("text/plain" | "text/*" | "*/*" | "*"), charset, _) =>
            new PlainTextSerializer(charset.getOrElse("utf-8"))
          case MessageProtocol(Some("application/json"), _, _) =>
            new JsonTextSerializer
        }
        .getOrElse {
          throw NotAcceptable(accepted, MessageProtocol(Some("text/plain")))
        }
  }

  /* ------ Deserializer for Client and Service -----*/
  def deserializer(protocol: MessageProtocol) = protocol.contentType match {
    case Some("text/plain") | None =>
      new PlainTextDeserializer(protocol.charset.getOrElse("utf-8"))
    case Some("application/json") =>
      new JsonTextDeserializer
    case _ =>
      // 抛出异常在生产环境并不可取,因为对于诸如WebSocket之类的应用,浏览器不允许设置ContentType
      // 此时应返回一个缺省的反序列化器更为妥当。
      throw UnsupportedMediaType(protocol, MessageProtocol(Some("text/plain")))
  }
}

另一个用Protocol buffers实现的例子:

import akka.util.ByteString
import com.lightbend.lagom.scaladsl.api.deser.MessageSerializer.NegotiatedDeserializer
import com.lightbend.lagom.scaladsl.api.deser.MessageSerializer.NegotiatedSerializer
import com.lightbend.lagom.scaladsl.api.deser.StrictMessageSerializer
import com.lightbend.lagom.scaladsl.api.transport.MessageProtocol

import scala.collection.immutable

class ProtobufSerializer extends StrictMessageSerializer[Order] {
  private final val serializer = {
    new NegotiatedSerializer[Order, ByteString]() {
      override def protocol: MessageProtocol =
        MessageProtocol(Some("application/octet-stream"))

      def serialize(order: Order) = {
        val builder = ByteString.createBuilder
        order.writeTo(builder.asOutputStream)
        builder.result
      }
    }
  }

  private final val deserializer = {
    new NegotiatedDeserializer[Order, ByteString] {
      override def deserialize(bytes: ByteString) =
        Order.parseFrom(bytes.iterator.asInputStream)
    }
  }

  override def serializerForRequest = serializer
  override def deserializer(protocol: MessageProtocol) = deserializer
  override def serializerForResponse(acceptedMessageProtocols: immutable.Seq[MessageProtocol]) = serializer
}

头部过滤器

在服务描述子中可以加入头部过滤器Header Filter,用于实现协商协议、身份验证或访问授权的沟通。过滤器会根据预设的条件,对服务与客户端双方通信的消息进行转换或修改。

下例是一个典型的过滤器实现。如果没有特意进行绑定,那么所有的服务默认都将会使用它,并使用ServicePrincipal来标识带有服务名称的客户端。在客户端,当Client发出Request时,过滤器将会在头部附加User-Agent,Lagom默认会自动将服务名称作为ServicePrinciple。而在服务端,则会读取Request中的User-Agent,并将其值设置为Request的Principle。

?? 切记:头部过滤器仅用于通信双方的协商,以确定双方采取何种方式进行后续的通信,而不应用来执行实际的验证逻辑。验证逻辑属于业务逻辑的组成部分,应当放在服务API及其组合当中。

object UserAgentHeaderFilter extends HeaderFilter {
  override def transformClientRequest(request: RequestHeader): RequestHeader = {
    request.principal match {
      case Some(principal: ServicePrincipal) =>
        request.withHeader(HeaderNames.USER_AGENT, principal.serviceName)
      case _ => request
    }
  }

  override def transformServerRequest(request: RequestHeader): RequestHeader = {
    request.getHeader(HeaderNames.USER_AGENT) match {
      case Some(userAgent) =>
        request.withPrincipal(ServicePrincipal.forServiceNamed(userAgent))
      case _ =>
        request
    }
  }

  override def transformServerResponse(response: ResponseHeader,request: RequestHeader): ResponseHeader = response

  override def transformClientResponse(response: ResponseHeader, request: RequestHeader): ResponseHeader = response
}

头部过滤器的组合

类似服务使用compose进行组合,头部过滤器也可以使用HeaderFilter.composite方法进行组合。

?? 注意:过滤器在发送消息与接收消息时适用的顺序是刚好相反的。对于Request,越后加入的过滤器越新鲜就越早被运用。

class VerboseFilter(name: String) extends HeaderFilter {
  private val log = LoggerFactory.getLogger(getClass)

  def transformClientRequest(request: RequestHeader) = {
    log.debug(name + " - transforming Client Request")
    request
  }

  def transformServerRequest(request: RequestHeader) = {
    log.debug(name + " - transforming Server Request")
    request
  }

  def transformServerResponse(response: ResponseHeader, request: RequestHeader) = {
    log.debug(name + " - transforming Server Response")
    response
  }

  def transformClientResponse(response: ResponseHeader, request: RequestHeader) = {
    log.debug(name + " - transforming Client Response")
    response
  }
}

/* ----- 按下列顺序组合过滤器 ----- */

def descriptor = {
  import Service._
  named("hello")
    .withCalls(
      call(sayHello)
    )
    .withHeaderFilter(
      HeaderFilter.composite(
        new VerboseFilter("Foo"),
        new VerboseFilter("Bar")
      )
    )
}

在服务端,控制台得到的输出将是如下的顺序:

[debug] Bar - transforming Server Request
[debug] Foo - transforming Server Request
[debug] Foo - transforming Server Response
[debug] Bar - transforming Server Response

错误的处理

Lagom在设计错误处理机制时,遵循了以下一些原则:

  • 在生产环境下,一个Lagom服务将永远不会将错误的细节暴露给另一个服务,除非知道这样做是肯定安全的。否则,未经审查和过滤的错误信息将有可能因为暴露了服务的内部细节,而被人利用。
  • 在开发环境下,则要尽可能地将异常的细节暴露出来,方便发现和定位Bug。
  • 如果可能,通常在客户端会复刻服务端触发的异常,以直接反馈服务发生失败的原因。
  • 如果可能,通常会将异常映射到HTTP的4字头或5字头的协议响应代码,或者是WebSocket的错误关闭代码。
  • 在断路器中实现HTTP响应代码映射,通常被视为最佳实践。

异常的序列化

Lagom为异常提供了trait ExceptionSerializer,用于将异常信息序列化为JSON等序列化格式,或者是特定的错误编码或响应代码。ExceptionSerializer会将异常转换为RawExceptionMessage,其中包括对应HTTP响应代码或WebSocket关闭代码的状态码、消息主体,以及一个关于协议的描述子(在HTTP,此处对应响应头部信息中的Content Type)。

默认的ExceptionSerializer使用Play JSON将异常信息序列化为JSON格式。除非是在开发模式下,否则它只会返回TransportException派生异常类的详细信息,最常见的派生类包括NotFoundPolicyViolation。Lagom通常也允许Client抛出这一类的异常,也允许自己创建或实现一个TransportException的派生类实例,不过前提是Client得认识这个异常并且知道如何进行反序列化。

额外的路由

从Lagom 1.5.0版本开始,允许使用Play Router对Lagom的服务进行扩展。该功能在需要将Lagom服务与既存的Play Router进行集成时显得更为实用。

在Lagom做DI时,可以注入额外的Router:

override lazy val lagomServer = serverFor[HelloService](wire[HelloServiceImpl])
  .additionalRouter(wire[SomePlayRouter])

一个关于文件上传的范例

此例基于ScalaSirdRouter实现,它将为服务添加一个/api/files的路径,用于接收支持断点续传数据的POST请求。

import play.api.mvc.DefaultActionBuilder
import play.api.mvc.PlayBodyParsers
import play.api.mvc.Results
import play.api.routing.Router
import play.api.routing.sird._

class FileUploadRouter(action: DefaultActionBuilder, parser: PlayBodyParsers) {
  val router = Router.from {
    case POST(p"/api/files") =>
      action(parser.multipartFormData) { request =>
        val filePaths = request.body.files.map(_.ref.getAbsolutePath)
        Results.Ok(filePaths.mkString("Uploaded[", ", ", "]"))
      }
  }
}

override lazy val lagomServer =
  serverFor[HelloService](wire[HelloServiceImpl])
    .additionalRouter(wire[FileUploadRouter].router)

在控制台使用命令curl -X POST -F "data=@somefile.txt" -v http://localhost:65499/api/files,即可发出上传请求。

与服务网关有关的注意事项

由于额外的路由并没有在服务描述子中定义,因此当服务使用了服务网关时,这些外部的路由将不会自动被服务网关暴露,因此需要显式地将它的路径加入访问控制列表ACL(Access Control List)里,然后通过服务网关访问它们。

trait HelloService extends Service {
  def hello(id: String): ServiceCall[NotUsed, String]

  final override def descriptor = {
    import Service._
    named("hello")
      .withCalls(
        pathCall("/api/hello/:id", hello _).withAutoAcl(true)
      )
      .withAcls(
        // extra ACL to expose additional router endpoint on ServiceGateway
        ServiceAcl(pathRegex = Some("/api/files"))
      )
  }
}

然后在控制台使用命令curl -X POST -F "data=@somefile.txt" -v http://localhost:9000/api/files访问它们。

与Lagom客户端有关的注意事项

由于额外的路由不是服务API的一部分,所以无法从Lagom生成的客户端进行直接的访问,而只能改用Play-WS之类的客户端去访问其暴露的HTTP端点。

编写可持久化和集群化的服务

这部分将使用Akka Typed按照DDD的方法实现一个CQRS架构的Lagom服务。

在这个ShoppingCart的示例里(?? 完整代码),使用了Dock作为容器,包装了Zookeeper、Kafka和PostGres服务,所以在演示前需要用docker-compose up -d进行初始化。同时,因为Lagom将使用Read-Side ProcessorTopic Producer对AggregateEventTag标记的Event进行消费,所以需要用AkkaTaggerAdapter.fromLagom把AggregateEventTag转换为Akka能理解的Tag类型。而在读端,Lagom提供的ReadSideProcessor,在Cassandra和Relational数据库插件支持下,可以为实现CQRS的读端提供完整的支持。

?? 使用JDBC驱动数据库存储Journal时,分片标记数不能超过10。这是该插件已知的一个Bug,如果超过了10,将会导致某些事件被多次传递。

1. 采用类似State Pattern的方式,把State与Handler设计在一起

/* ----- State & Handlers ----- */
final case class ShoppingCart(
    items: Map[String, Int],
    // checkedOutTime defines if cart was checked-out or not:
    // case None, cart is open
    // case Some, cart is checked-out
    checkedOutTime: Option[Instant] = None){
  /* ----- Command Handlers ----- */
  def applyCommand(cmd: ShoppingCartCommand): ReplyEffect[ShoppingCartEvent, ShoppingCart] =
    if (isOpen) {
      cmd match {
        case AddItem(itemId, quantity, replyTo) => onAddItem(itemId, quantity, replyTo)
        case Checkout(replyTo)                  => onCheckout(replyTo)
        case Get(replyTo)                       => onGet(replyTo)
      }
    } else {
      cmd match {
        case AddItem(_, _, replyTo) => Effect.reply(replyTo)(Rejected("Cannot add an item to a checked-out cart"))
        case Checkout(replyTo)      => Effect.reply(replyTo)(Rejected("Cannot checkout a checked-out cart"))
        case Get(replyTo)           => onGet(replyTo)
      }
    }

  private def onAddItem(itemId: String, quantity: Int, replyTo: ActorRef[Confirmation]): ReplyEffect[ShoppingCartEvent, ShoppingCart] = {
    if (items.contains(itemId))
      Effect.reply(replyTo)(Rejected(s"Item ‘$itemId‘ was already added to this shopping cart"))
    else if (quantity <= 0)
      Effect.reply(replyTo)(Rejected("Quantity must be greater than zero"))
    else
      Effect
        .persist(ItemAdded(itemId, quantity))
        .thenReply(replyTo)(updatedCart => Accepted(toSummary(updatedCart)))
  }

  private def onCheckout(replyTo: ActorRef[Confirmation]): ReplyEffect[ShoppingCartEvent, ShoppingCart] = {
    if (items.isEmpty)
      Effect.reply(replyTo)(Rejected("Cannot checkout an empty shopping cart"))
    else
      Effect
        .persist(CartCheckedOut(Instant.now()))
        .thenReply(replyTo)(updatedCart => Accepted(toSummary(updatedCart)))
  }

  private def onGet(replyTo: ActorRef[Summary]): ReplyEffect[ShoppingCartEvent, ShoppingCart] = {
    Effect.reply(replyTo)(toSummary(shoppingCart = this))
  }

  private def toSummary(shoppingCart: ShoppingCart): Summary = {
    Summary(shoppingCart.items, shoppingCart.checkedOut)
  }

  /* ----- Event Handlers ----- */
  def applyEvent(evt: ShoppingCartEvent): ShoppingCart =
    evt match {
      case ItemAdded(itemId, quantity)    => onItemAdded(itemId, quantity)
      case CartCheckedOut(checkedOutTime) => onCartCheckedOut(checkedOutTime)
    }

  private def onItemAdded(itemId: String, quantity: Int): ShoppingCart =
    copy(items = items + (itemId -> quantity))

  private def onCartCheckedOut(checkedOutTime: Instant): ShoppingCart = {
    copy(checkedOutTime = Option(checkedOutTime))
  }
}

2. 将Protocol与Factory放在Companion Object里

Lagom借用Akka Cluster Sharding实现了服务的集群部署,确保在任意时刻,有且只有一个聚合的实例在集群内活动,相同类型的多个聚合实例则均匀分布在各个节点之上。为此,需要给ShoppingCart设定一个EntityKey,并向其工厂方法传入EntityContext。

?? 在Command-Reply-Event的设计上,要注意区别Reply是回复调用者的副作用,它可以是确认Command是否成功执行,或者是返回聚合的内部状态(要注意区别查询命令与Read-Side)。而Event才是因Command而导致聚合发生变化的正作用,是要持久化的。相应的,Effect.Reply()用于聚合未发生变化的场合,而Effect.persist().thenReply()则用于聚合发生变化之后。

/* ----- Factory & Protocols ----- */
object ShoppingCart {
  val empty                                       = ShoppingCart(items = Map.empty)
  val typeKey: EntityTypeKey[ShoppingCartCommand] = EntityTypeKey[ShoppingCartCommand]("ShoppingCart")

  /* ----- Commands ----- */
  trait CommandSerializable

  sealed trait ShoppingCartCommand extends CommandSerializable

  final case class AddItem(itemId: String, quantity: Int, replyTo: ActorRef[Confirmation])
    extends ShoppingCartCommand

  final case class Checkout(replyTo: ActorRef[Confirmation]) extends ShoppingCartCommand

  final case class Get(replyTo: ActorRef[Summary]) extends ShoppingCartCommand

  /* ----- Replies (will not be persisted) ----- */
  sealed trait Confirmation

  final case class Accepted(summary: Summary) extends Confirmation

  final case class Rejected(reason: String) extends Confirmation

  final case class Summary(items: Map[String, Int], checkedOut: Boolean)

  /* ----- Events (will be persisted) ----- */
  sealed trait ShoppingCartEvent extends AggregateEvent[ShoppingCartEvent] {
    override def aggregateTag: AggregateEventTagger[ShoppingCartEvent] = ShoppingCartEvent.Tag
  }

  final case class ItemAdded(itemId: String, quantity: Int) extends ShoppingCartEvent

  final case class CartCheckedOut(eventTime: Instant) extends ShoppingCartEvent

  /* ----- Tag for read-side consuming ----- */
  object ShoppingCartEvent {
    // will produce tags with shard numbers from 0 to 9
    val Tag: AggregateEventShards[ShoppingCartEvent] =
      AggregateEventTag.sharded[ShoppingCartEvent](numShards = 10)
  }

  def apply(entityContext: EntityContext[ShoppingCartCommand]): Behavior[ShoppingCartCommand] = {
    EventSourcedBehavior
      .withEnforcedReplies[ShoppingCartCommand, ShoppingCartEvent, ShoppingCart](
        persistenceId = PersistenceId(entityContext.entityTypeKey.name, entityContext.entityId),
        emptyState = ShoppingCart.empty,
        commandHandler = (cart, cmd) => cart.applyCommand(cmd),
        eventHandler = (cart, evt) => cart.applyEvent(evt)
      )
      // convert tag of Lagom to tag of Akka
      .withTagger(AkkaTaggerAdapter.fromLagom(entityContext, ShoppingCartEvent.Tag))
      // snapshot every 100 events and keep at most 2 snapshots on db
      .withRetention(RetentionCriteria.snapshotEvery(numberOfEvents = 100, keepNSnapshots = 2))
  }
}

3. 实现Lagom用于启动服务的ApplicationLoader,集成形成完整应用

class ShoppingCartLoader extends LagomApplicationLoader {
  override def load(context: LagomApplicationContext): LagomApplication =
    new ShoppingCartApplication(context) with AkkaDiscoveryComponents

  override def loadDevMode(context: LagomApplicationContext): LagomApplication =
    new ShoppingCartApplication(context) with LagomDevModeComponents

  override def describeService = Some(readDescriptor[ShoppingCartService])
}

trait ShoppingCartComponents
    extends LagomServerComponents
    with SlickPersistenceComponents
    with HikariCPComponents
    with AhcWSComponents {
  implicit def executionContext: ExecutionContext

  override lazy val lagomServer: LagomServer =
    serverFor[ShoppingCartService](wire[ShoppingCartServiceImpl])
  override lazy val jsonSerializerRegistry: JsonSerializerRegistry =
    ShoppingCartSerializerRegistry

  // Initialize the sharding for the ShoppingCart aggregate.
  // See https://doc.akka.io/docs/akka/2.6/typed/cluster-sharding.html
  clusterSharding.init(
    Entity(ShoppingCart.typeKey) { entityContext =>
      ShoppingCart(entityContext)
    }
  )
}

abstract class ShoppingCartApplication(context: LagomApplicationContext)
    extends LagomApplication(context)
    with ShoppingCartComponents
    with LagomKafkaComponents {}

4. 在服务的实现中在服务的实现中使用Ask模式访问Actor

服务API的相关操作实际将由幕后的Actor负责实现,因此在服务的实现里需要访问Actor的实例,为此需要通过ClusterSharding.entityRefFor获取其EntityRef。

class ShoppingCartServiceImpl(
    clusterSharding: ClusterSharding,
    persistentEntityRegistry: PersistentEntityRegistry
  )(implicit ec: ExecutionContext) extends ShoppingCartService {
  def entityRef(id: String): EntityRef[ShoppingCartCommand] =
    clusterSharding.entityRefFor(ShoppingCart.typeKey, id)
}

在获得Reference后,便可使用Ask模式与Actor进行交互。Ask返回的是Future[Response],因此示例将Future[Summary]投射为ShoppingCartView,方便Read-Side使用。

implicit val timeout = Timeout(5.seconds)

override def get(id: String): ServiceCall[NotUsed, ShoppingCartView] = ServiceCall { _ =>
  entityRef(id)
    .ask(reply => Get(reply))
    .map(cartSummary => asShoppingCartView(id, cartSummary))
}

final case class ShoppingCartItem(itemId: String, quantity: Int)
final case class ShoppingCartView(id: String, items: Seq[ShoppingCartItem], checkedOut: Boolean)

private def asShoppingCartView(id: String, cartSummary: Summary): ShoppingCartView = {
  ShoppingCartView(
    id,
    cartSummary.items.map((ShoppingCartItem.apply _).tupled).toSeq,
    cartSummary.checkedOut
  )
}

5. 其他一些需要考虑的细节

  • 分片数与节点数的协调:分片数少于节点数,某些节点将无所事事;分片数多于节点数,节点的工作量需要细心平衡。
  • 实体钝化:让所有的聚合实例都始终活在内存里并不是高效的办法。对较长时间内没有活动的聚合,可以使用实体钝化功能将其暂时从分片节点上移除,需要它时再重新唤醒并加载即可。
  • 数据的序列化:选择JSON、二进制串等格式进行序列化时,需要注意的是消息中可能包含ActorRef这样的字段,所以这部分内容要参考Akka指南。

迁移到Akka Persistence Typed

这部分内容是对比上一节直接使用Akka Persistence Typed建立领域模型的方式,改从传统的Lagom Persistence迁移到Akka Persistence Typed的角度进行了详细的分步讲解。所以,如果是全新开始设计的Lagom的服务,建议直接使用Akka Persistence Typed进行实现,只有此前用Lagom Persistence实现的服务才需要考虑迁移。

由于内容主要涉及Akka Typed,可参考我的博客内容:

选择合适的数据库平台

Lagom与下列数据库平台兼容:

  • Cassandra
  • PostgreSQL
  • MySQL
  • Oracle
  • H2
  • Microsoft SQL Server
  • Couchbase

参考链接:

Cassandra

Cassandra需要至少3个KeySpace:

  • Journal:存储Event:cassandra-journal.keyspace = my_service_journal
  • Snapshot:存储快照:cassandra-snapshot-store.keyspace = my_service_snapshot
  • Offset:存储Read-Side侧最近处理的Event:lagom.persistence.read-side.cassandra.keyspace = my_service_read_side

这只是Lagom官方文档的一小部分内容,算是对如何使用该框架实现服务的初窥,有兴趣的请移步官方网站寻找更多的内容。

Lagom 官方文档之随手记

标签:sem   地方   latest   discovery   root   代码   ber   catch   rod   

原文地址:https://www.cnblogs.com/Abbey/p/13380265.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!