akka-typed(9) - 業務分片、整合，談談lagom, 需要嗎？
在討論lagom之前，先從遇到的需求開始介紹：現代企業的it系統變得越來越多元化、複雜化了。線上、線下各種系統必須用某種方式整合在一起。從各種it系統的基本共性分析：最明顯的特徵應該是後臺資料庫的角色了，起碼，大家都需要使用資料。另外，每個系統都可能具備大量實時線上使用者、海量資料特性，代表著對資料處理能力有極大的要求，預示系統只有通過分散式處理方式才能有效執行。
一個月前開始設計一個企業的it系統，在討論資料中臺時就遇到這樣的需求。這個所謂的資料中臺的主要作用是為整體系統提供一套統一的資料使用api，前後連線包括web,mobile,desktop的前端系統以及由多種傳統及分散式資料庫系統，形成一個統一的資料使用介面。實際上，資料庫連線不只是簡單的讀寫操作，還需要包括所有實時的資料處理：根據業務要求對資料進行相應的處理然後使用。那麼這是一個怎樣的系統呢？首先，它必須是分散式的：為了對付大量的前端使用者同時呼叫同一個api，把這個api的功能同時分派到多個伺服器上執行是個有效的解決方法。這是個akka-cluster-sharding模式。資料中臺api是向所有內部系統以及一些特定的外部第三方系統開放的，用http標準協議支援各系統與資料後臺的連線也是合理的。這個akka-http, akka-grpc可以勝任。然後各系統之間的整合可以通過一個流運算工具如kafka實現各聚合根之間的互動連線。
似乎所有需要的工具都齊備了，其中akka佔了大部分功能。但有些問題是：基於akka技術棧來程式設計或多或少有些門檻要求。最起碼需要一定程度的akka開發經驗。更不用提組織一個開發團隊了。如果市面上有個什麼能提供相應能力的開發工具，可以輕鬆快速上手的，那麼專案開發就可以立即啟動了。
現在來談談lagom：lagom是一套scala棧的微服務軟體開發工具。從官方文件介紹瞭解到lagom主要提供了一套服務介面定義及服務功能開發框架。值得一提的是服務功能可以是叢集分片模式的。走了一遍lagom的啟動示範程式碼，感覺這是一套集開發、測試、部署為一體的框架（framework）。在這個框架裡按照規定開發幾個簡單的服務api非常順利，很方便。這讓我對使用lagom產生了興趣，想繼續調研一下利用lagoom來開發上面所提及資料中臺的可行性。lagom服務接入部分是通過play實現的。play我不太熟悉，想深入瞭解一下用akka-http替代的可行性，不過看來不太容易。最讓我感到失望的是lagom的服務分片（service-sharding）直接就是akka-cluster那一套：cluster、event-sourcing、CQRS什麼的都需要自己從頭到尾重新編寫。用嵌入的kafka進行服務整合與單獨用kafka也不會增加太多麻煩。倒是lagom提供的這個集開發、測試、部署為一體的框架在團隊開發管理中應該能發揮良好的作用。
在我看來：服務接入方面由於涉及身份驗證、使用許可權、二進位制檔案型別資料交換等使用akka-http,akka-grpc會更有控制力。服務功能實現直接就用akka-cluster-sharding，把計算任務分佈到各節點上，這個我們前面已經介紹過了。
所以，最後還是決定直接用akka-typed來實現這個資料中臺。用了一個多月時間做研發，到現在看來效果不錯，能夠符合專案要求。下面是一些用akka-typed實現業務整合的過程介紹。首先，系統特點是功能分片：系統按業務條塊分成多個片shardregion，每個片裡的entity負責處理一項業務的多個功能。多個使用者呼叫一項業務功能代表多個entity分佈在不同的叢集節點上並行運算。下面是一個業務群的程式碼示範：
object Shards extends LogSupport {
def apply(mgoHosts: List[String],trace: Boolean, keepAlive: FiniteDuration, pocurl: String)(
implicit authBase: AuthBase): Behavior[Nothing] = {
Behaviors.setup[Nothing] { ctx =>
val sharding = ClusterSharding(ctx.system)
log.stepOn = true
log.step(s"starting cluster-monitor ...")(MachineId("",""))
ctx.spawn(MonitorActor(),"abs-cluster-monitor")
log.step(s"initializing sharding for ${Authenticator.EntityKey} ...")(MachineId("",""))
val authEntityType = Entity(Authenticator.EntityKey) { entityContext =>
Authenticator(entityContext.shard,mgoHosts,trace,keepAlive)
}.withStopMessage(Authenticator.StopAuthenticator)
sharding.init(authEntityType)
log.step(s"initializing sharding for ${CrmWorker.EntityKey} ...")(MachineId("",""))
val crmEntityType = Entity(CrmWorker.EntityKey) { entityContext =>
CrmWorker(entityContext.shard,mgoHosts,entityContext.entityId,trace,keepAlive)
}.withStopMessage(CrmWorker.StopWorker)
sharding.init(crmEntityType)
log.step(s"initializing sharding for ${GateKeeper.EntityKey} ...")(MachineId("",""))
val gateEntityType = Entity(GateKeeper.EntityKey) { entityContext =>
GateKeeper(entityContext.shard,mgoHosts,entityContext.entityId,trace,keepAlive)
}.withStopMessage(GateKeeper.StopGateKeeper)
sharding.init(gateEntityType)
log.step(s"initializing sharding for ${PluWorker.EntityKey} ...")(MachineId("",""))
val pluEntityType = Entity(PluWorker.EntityKey) { entityContext =>
PluWorker(entityContext.shard,mgoHosts,entityContext.entityId,trace,keepAlive)
}.withStopMessage(PluWorker.StopWorker)
sharding.init(pluEntityType)
log.step(s"initializing sharding for ${PocWorker.EntityKey} ...")(MachineId("",""))
val pocEntityType = Entity(PocWorker.EntityKey) { entityContext =>
PocWorker(entityContext.shard,mgoHosts,entityContext.entityId,trace,keepAlive,pocurl)
}.withStopMessage(PocWorker.StopWorker)
sharding.init(pocEntityType)
Behaviors.empty
}
}
}
可以看到，不同型別的片以不同的EntityKey來代表。前端接入是基於akka-http的，如下：
object CrmRoute extends LogSupport {
def route(entityRef: EntityRef[CrmWorker.Command])(
implicit ec: ExecutionContext, jsStreaming: EntityStreamingSupport, timeout: Timeout): akka.http.scaladsl.server.Route = {
concat(
pathPrefix("ismember") {
parameter(Symbol("faceid")) { fid =>
val futResp = entityRef.ask[CrmWorker.Response](CrmWorker.IsMemberFace(fid, _))
.map {
case CrmWorker.ValidMember(memberId) => memberId
case CrmWorker.InvalidMember(msg) => throw new Exception(msg)
}
onSuccess(futResp)(complete(_))
}
},
pathPrefix("getmember") {
parameter(Symbol("memberid")) { mid =>
val futResp = entityRef.ask[CrmWorker.Response](CrmWorker.GetMemberInfo(mid, _))
.map {
case CrmWorker.MemberInfo(json) => HttpEntity(MediaTypes.`application/json`,json)
case CrmWorker.InvalidMemberInfo(msg) => throw new Exception(msg)
}
onSuccess(futResp)(complete(_))
}
}
)
}
}
各項業務功能呼叫通過entityRef.ask傳送給了某個使用者指定節點上的entity。akka的actor是執行緒的再細分，即一個actor可能與其它成千上萬個actor共享一條執行緒。所以絕對不容許任何blocking。我是用下面示範的模式來實現non-blocking的：
def apply(shard: ActorRef[ClusterSharding.ShardCommand],mgoHosts: List[String], entityId: String, trace: Boolean, keepAlive: FiniteDuration): Behavior[Command] = {
val (shopId,posId) = entityId.split(':').toList match {
case sid::pid::Nil => (sid,pid) }
implicit val loc = Messages.MachineId(shopId,posId)
log.stepOn = trace
// Behaviors.supervise(
Behaviors.setup[Command] { ctx =>
implicit val ec = ctx.executionContext
ctx.setReceiveTimeout(keepAlive, Idle)
Behaviors.withTimers[Command] { timer =>
Behaviors.receiveMessage[Command] {
case IsMemberFace(fid, replyTo) =>
log.step(s"CrmWorker: IsMemberFace($fid)")
implicit val client = mongoClient(mgoHosts)
maybeMgoClient = Some(client)
ctx.pipeToSelf(isMemberFace(fid)) {
case Success(mid) => {
if (mid._1.isEmpty) {
replyTo ! InvalidMember(mid._2)
Done(loc.shopid, loc.posid, s"IsMemberFace with Error ${mid._2}")
} else {
replyTo ! ValidMember(mid._1)
Done(loc.shopid, loc.posid, s"IsMemberFace.")
}
}
case Failure(err) =>
log.error(s"CrmWorker: IsMemberFace Error: ${err.getMessage}")
replyTo ! InvalidMember(err.getMessage)
Done(loc.shopid, loc.posid, s"IsMemberFace with error: ${err.getMessage}")
}
Behaviors.same
case GetMemberInfo(mid, replyTo) =>
log.step(s"CrmWorker: GetMemberInfo($mid)")
implicit val client = mongoClient(mgoHosts)
maybeMgoClient = Some(client)
ctx.pipeToSelf(getMemberInfo(mid)) {
case Success(json) => {
replyTo ! MemberInfo(json)
Done(loc.shopid, loc.posid, s"GetMemberInfo with json ${json}")
}
case Failure(err) =>
log.error(s"CrmWorker: GetMemberInfo Error: ${err.getMessage}")
replyTo ! InvalidMemberInfo(err.getMessage)
Done(loc.shopid, loc.posid, s"GetMemberInfo with error: ${err.getMessage}")
}
Behaviors.same
case Idle =>
// after receive timeout
shard ! ClusterSharding.Passivate(ctx.self)
Behaviors.same
case StopWorker =>
Behaviors.stopped(
() => log.step(s"CrmWorker: {$shopId,$posId} passivated to stop.")(MachineId(shopId, posId))
)
case Done(shopid, termid, work) =>
if (maybeMgoClient.isDefined)
maybeMgoClient.get.close()
log.step(s"CrmWorker: {$shopid,$termid} finished $work")(MachineId(shopid,termid))
Behaviors.same
case _ => Behaviors.same
}.receiveSignal {
case (_,PostStop) =>
log.step(s"CrmWorker: {$shopId,$posId} stopped.")(MachineId(shopId, posId))
Behaviors.same
}
}
}
// ).onFailure(SupervisorStrategy.restart)
}
主要是使用ctx.pipeToSelf(work)把一個Future轉換成內部訊息。這裡的work的實現最終必須返回Future型別，如下面的示範：
object CrmServices extends JsonConverter with LogSupport {
import MgoHelpers._
def validMember(docs: Seq[Document], faceid: String): Future[(String,String)] = {
val memberId: (String, String) = docs match {
case Nil => ("", s"faceid[$faceid]不存在！")
case docs =>
val member = MemberInfo.fromDocument(docs.head)
if (member.expireDt.compareTo(mgoDateTimeNow) < 0)
("", s"會員：${member.memberId}-${member.memberName}會籍已過期！")
else
(member.memberId, "")
}
FastFuture.successful(memberId)
}
def isMemberFace(faceid: String)(
implicit mgoClient: MongoClient, ec: ExecutionContext): Future[(String,String)] = {
implicit val db = mgoClient.getDatabase(CrmModels.SCHEMA.DBNAME)
val col = db.getCollection(CrmModels.SCHEMA.MEMBERINFO)
val memberInfo: Future[Seq[Document]] = col.find(equal(SCHEMA.FACEID,faceid)).toFuture()
for {
mi <- memberInfo
(id,msg) <- validMember(mi,faceid)
} yield (id,msg)
}
def getMemberInfo(memberid: String)(
implicit mgoClient: MongoClient, ec: ExecutionContext): Future[String] = {
implicit val db = mgoClient.getDatabase(CrmModels.SCHEMA.DBNAME)
val col = db.getCollection(CrmModels.SCHEMA.MEMBERINFO)
val memberInfo: Future[Seq[Document]] = col.find(equal(SCHEMA.MEMBERID,memberid)).toFuture()
for {
docs <- memberInfo
jstr <- FastFuture.successful(if(docs.isEmpty) "" else toJson(MemberInfo.fromDocument(docs.head)))
} yield jstr
}
}
另外，由於每個使用者第一次呼叫一項業務功能時akka-cluster-shardregion都會自動在某個節點上構建一個新的entity，如果上萬個使用者使用過某個功能，那麼就會有萬個entity及其所佔用的資源如mongodb客戶端等停留在記憶體裡。所以在完成一項功能運算後應關閉entity，釋放佔用的資源。這個是通過shard ! ClusterSharding.passivate(ctx.self)實現的。