akka-typed(8) - CQRS讀寫分離模式-有解無憂

前面介紹了事件源（EventSource）和集群（cluster），現在到了討論CQRS的時候了，CQRS即讀寫分離模式，由獨立的寫方程式和讀方程式組成，具體原理在以前的博客里介紹過了，akka-typed應該自然支持CQRS模式，最起碼本身提供了對寫方編程的支持，這點從EventSourcedBehavior 可以知道，akka-typed提供了新的EventSourcedBehavior-Actor，極大方便了對persistentActor的應用開發，但同時也給編程者造成了一些限制，如手工改變狀態會更困難了、EventSourcedBehavior不支持多層式的persist，也就是說通過persist某些特定的event然后在event-handler程式里進行狀態處理是不可能的了，我這里有個例子，是個購物車應用：當完成支付后需要取個快照(snapshot)，下面是這個snapshot的代碼：

       snapshotWhen {
          (state,evt,seqNr) => CommandHandler.takeSnapshot(state,evt,seqNr)
       }
...
 
  def takeSnapshot(state: Voucher, evt: Events.Action, lstSeqNr: Long)(implicit pid: PID) = {
    if (evt.isInstanceOf[Events.PaymentMade]
        || evt.isInstanceOf[Events.VoidVoucher.type]
        || evt.isInstanceOf[Events.SuspVoucher.type])
      if (state.items.isEmpty) {
        log.step(s"#${state.header.num} taking snapshot at [$lstSeqNr] ...")
        true
      } else
        false
    else
      false


  }

判斷event型別是沒有問題的，因為正是當前的事件，但另一個條件是購物車必須是清空了的，這個有點為難，因為這個狀態要依賴這幾個event運算的結果才能確定，也就是下一步，但確定結果又需要對購物車內容進行計算，好像是個死回圈，在akka-classic里我們可以在判斷了event運算結果后，如果需要改變狀態就再persist一個特殊的event，然后在這個event的handler進行狀態處理，沒辦法，EventSourcedBehavior不支持多層persist，只有這樣做：

      case PaymentMade(acct, dpt, num, ref,amount) =>
             ...
              writerInternal.lastVoucher = Voucher(vchs, vItems)
              endVoucher(Voucher(vchs,vItems),TXNTYPE.sales)
              Voucher(vchs.nextVoucher, List())
             ...

我只能先吧當前狀態保存下來、進行結單運算、然后清空購物車，這樣snapshot就可以順利進行了，

好了，akka的讀方編程是通過PersistentQuery實作的，reader的作用就是把event從資料庫讀出來后再恢復成具體的資料格式，我們從reader的呼叫了解一下這個應用里reader的實作細節：

    val readerShard = writerInternal.optSharding.get   
    val readerRef = readerShard.entityRefFor(POSReader.EntityKey, s"$pid.shopId:$pid.posId")
    readerRef ! Messages.PerformRead(pid.shopid, pid.posid,writerInternal.lastVoucher.header.num,writerInternal.lastVoucher.header.opr,bseq,eseq,txntype,writerInternal.expurl,writerInternal.expacct,writerInternal.exppass)

可以看到這個reader是一個集群分片，sharding-entity，想法是每單完成購買后發個訊息給一個entity、這個entity再完成reader功能后自動終止，立即釋放出占用的資源，reader-actor的定義如下：

object POSReader extends LogSupport {
  val EntityKey: EntityTypeKey[Command] = EntityTypeKey[Command]("POSReader")

  def apply(nodeAddress: String, trace: Boolean): Behavior[Command] = {
    log.stepOn = trace
    implicit var pid: PID = PID("","")
    Behaviors.supervise(
      Behaviors.setup[Command] { ctx =>
        Behaviors.withTimers { timer =>
          implicit val ec = ctx.executionContext
          Behaviors.receiveMessage {
            case PerformRead(shopid, posid, vchnum, opr, bseq, eseq, txntype, xurl, xacct, xpass) =>
              pid = PID(shopid, posid)
              log.step(s"POSReader: PerformRead($shopid,$posid,$vchnum,$opr,$bseq,$eseq,$txntype,$xurl,$xacct,$xpass)")(PID(shopid, posid))
              val futReadSaveNExport = for {
                txnitems <- ActionReader.readActions(ctx, vchnum, opr, bseq, eseq, trace, nodeAddress, shopid, posid, txntype)
                _ <- ExportTxns.exportTxns(xurl, xacct, xpass, vchnum, txntype == Events.TXNTYPE.suspend,
                     { if(txntype == Events.TXNTYPE.voidall)
                       txnitems.map (_.copy(txntype=Events.TXNTYPE.voidall))
                     else txnitems },
                     trace)(ctx.system.toClassic, pid)
              } yield ()
              ctx.pipeToSelf(futReadSaveNExport) {
                case Success(_) => {
                  timer.startSingleTimer(ReaderFinish(shopid, posid, vchnum), readInterval.seconds)
                  StopReader
                }
                case Failure(err) =>
                  log.error(s"POSReader:  Error: ${err.getMessage}")
                  timer.startSingleTimer(ReaderFinish(shopid, posid, vchnum), readInterval.seconds)
                  StopReader
              }

            Behaviors.same
            case StopReader =>
              Behaviors.same
            case ReaderFinish(shopid, posid, vchnum) =>
              Behaviors.stopped(
                () => log.step(s"POSReader: {$shopid,$posid} finish reading voucher#$vchnum and stopped")(PID(shopid, posid))
              )
          }
        }
      }
    ).onFailure(SupervisorStrategy.restart)
  }

reader就是一個普通的actor，值得注意的是讀方程式可能是一個龐大復雜的程式，肯定需要分割成多個模塊，所以我們可以按照流程順序進行模塊功能切分：這樣下面的模塊可能會需要上面模塊產生的結果才能繼續，記住，在actor中絕對避免阻塞執行緒，所有的模塊都回傳Future, 然后用for-yield串起來，上面我們用了ctx.pipeToSelf 在Future運算完成后發送ReaderFinish訊息給自己，通知自己停止，

在這個例子里我們把reader任務分成：

1、從資料庫讀取事件

2、事件重演一次產生狀態資料（購物車內容）

3、將形成的購物車內容作為交易單據專案存入資料庫

4、向用戶提供的restapi輸出交易資料

event讀取是通過cassandra-persistence-plugin實作的：

    val query =
    PersistenceQuery(classicSystem).readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)

    // issue query to journal
    val source: Source[EventEnvelope, NotUsed] =
      query.currentEventsByPersistenceId(s"${pid.shopid}:${pid.posid}", startSeq, endSeq)

    // materialize stream, consuming events
    val readActions: Future[List[Any]] = source.runFold(List[Any]()) { (lstAny, evl) => evl.event :: lstAny }

這部分比較簡單：定義一個PersistenceQuery，用它產生一個Source，然后run這個Source獲取Future[List[Any]]，

重演事件產生交易資料：

    def buildVoucher(actions: List[Any]): List[TxnItem] = {
      log.step(s"POSReader: read actions: $actions")
      val (voidtxns,onlytxns) = actions.asInstanceOf[Seq[Action]].pickOut(_.isInstanceOf[Voided])
      val listOfActions = onlytxns.reverse zip (LazyList from 1)   //zipWithIndex
      listOfActions.foreach { case (txn,idx) =>
        txn.asInstanceOf[Action] match {
          case Voided(_) =>
          case ti@_ =>
            curTxnItem = EventHandlers.buildTxnItem(ti.asInstanceOf[Action],vchState).copy(opr=cshr)
            if(voidtxns.exists(a => a.asInstanceOf[Voided].seq == idx)) {
              curTxnItem = curTxnItem.copy(txntype = TXNTYPE.voided, opr=cshr)
              log.step(s"POSReader: voided txnitem: $curTxnItem")
            }
            val vch = EventHandlers.updateState(ti.asInstanceOf[Action],vchState,vchItems,curTxnItem,true)
            vchState = vch.header
            vchItems = vch.txnItems
            log.step(s"POSReader: built txnitem: ${vchItems.txnitems.head}")
        }
      }
      log.step(s"POSReader: voucher built with state: $vchState, items: ${vchItems.txnitems}")
      vchItems.txnitems
    }

重演List[Event]，產生了List[TxnItem]，

向資料庫里寫List[TxnItem]：

 def writeTxnsToDB(vchnum: Int, txntype: Int, bseq: Long, eseq: Long, txns: List[TxnItem])(
                   implicit system: akka.actor.ActorSystem, session: CassandraSession, pid: PID): Future[Seq[TxnItem]] = ???

注意回傳結果型別Future[Seq[TxnItem]]，我們用for-yield把這幾個動作串起來：

  val txnitems: Future[List[Events.TxnItem]] = for {
      lst1 <- readActions    //read list from Source
      lstTxns <- if (lst1.length < (endSeq -startSeq))    //if imcomplete list read again
        readActions
        else FastFuture.successful(lst1)
      items <- FastFuture.successful( buildVoucher(lstTxns) )
      _ <- JournalTxns.writeTxnsToDB(vchnum,txntype,startSeq,endSeq,items)
      _ <- session.close(ec)
    } yield items

注意回傳結果型別Future[Seq[TxnItem]]，我們用for-yield把這幾個動作串起來：

  val txnitems: Future[List[Events.TxnItem]] = for {
      lst1 <- readActions    //read list from Source
      lstTxns <- if (lst1.length < (endSeq -startSeq))    //if imcomplete list read again
        readActions
        else FastFuture.successful(lst1)
      items <- FastFuture.successful( buildVoucher(lstTxns) )
      _ <- JournalTxns.writeTxnsToDB(vchnum,txntype,startSeq,endSeq,items)
      _ <- session.close(ec)
    } yield items

注意：這個for回傳的Future[List[TxnItem]]，是提供給restapi輸出功能的，在那里List[TxnItem]會被轉換成json作為post的包嵌資料，

現在所有子任務的回傳結果型別都是Future了，我們可以再用for來把它們串起來：

             val futReadSaveNExport = for {
                txnitems <- ActionReader.readActions(ctx, vchnum, opr, bseq, eseq, trace, nodeAddress, shopid, posid, txntype)
                _ <- ExportTxns.exportTxns(xurl, xacct, xpass, vchnum, txntype == Events.TXNTYPE.suspend,
                     { if(txntype == Events.TXNTYPE.voidall)
                       txnitems.map (_.copy(txntype=Events.TXNTYPE.voidall))
                     else txnitems },
                     trace)(ctx.system.toClassic, pid)
              } yield ()

說到EventSourcedBehavior，因為用了cassandra-plugin，忽然想起組態檔里新舊有很大區別，現在這個application.conf是這樣的：

akka {
  loglevel = INFO
  actor {
    provider = cluster
    serialization-bindings {
      "com.datatech.pos.cloud.CborSerializable" = jackson-cbor
    }
  }
  remote {
    artery {
      canonical.hostname = "192.168.11.189"
      canonical.port = 0
    }
  }
  cluster {
    seed-nodes = [
      "akka://[email protected]:2551"]
    sharding {
      passivate-idle-entity-after = 5 m
    }
  }
  # use Cassandra to store both snapshots and the events of the persistent actors
  persistence {
    journal.plugin = "akka.persistence.cassandra.journal"
    snapshot-store.plugin = "akka.persistence.cassandra.snapshot"
  }
}
akka.persistence.cassandra {
  # don't use autocreate in production
  journal.keyspace = "poc2g"
  journal.keyspace-autocreate = on
  journal.tables-autocreate = on
  snapshot.keyspace = "poc2g_snapshot"
  snapshot.keyspace-autocreate = on
  snapshot.tables-autocreate = on
}

datastax-java-driver {
  basic.contact-points = ["192.168.11.189:9042"]
  basic.load-balancing-policy.local-datacenter = "datacenter1"
}

akka.persitence.cassandra段落里可以定義keyspace名稱，這樣新舊版本應用可以共用一個cassandra,同時在線，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/422.html

標籤：Scala

上一篇：akka-typed(7) - cluster：sharding, 集群分片

下一篇：Scala 中的約定