如何将域模型设置为参与者？答案

【问题标题】：How to setup domain model as actor?如何将域模型设置为参与者？
【发布时间】：2019-05-31 22:11:04
【问题描述】：

我对 Scala 和 Akka 都很陌生，我正在尝试弄清楚如何创建一个合适的域模型，它也是一个 Actor。

假设我们有一个简单的商业案例，您可以在其中开设一个新的银行账户。假设其中一条规则是每个姓氏只能创建一个银行账户（不现实，只是为了简单起见）。我的第一种方法，不应用任何业务规则，看起来像这样：

object Main {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("accout")
    implicit val materializer = ActorMaterializer()
    implicit val executionContext = system.dispatcher
    val account = system.actorOf(Props[Account])
    account ! CreateAccount("Doe")
  }
}

case class CreateAccount(lastName: String)

class Account extends Actor {

  var lastName: String = null

  override def receive: Receive = {
    case createAccount: CreateAccount =>
      this.lastName = lastName
  }
}

最终，您会将这些数据保存在某个地方。但是，当添加一个姓氏只能有一个银行账户的规则时，需要对一些数据存储进行查询。假设我们将该逻辑放在存储库中，并且存储库最终返回 Account，我们遇到了Account 不再是 Actor 的问题，因为存储库将无法创建 Actor。

这绝对是一个错误的实现，而不是应该如何使用 Actors。我的问题是，有什么方法可以解决这些问题？我知道我对 Akka 的了解还不够高，所以这可能是一个奇怪/愚蠢的问题。

【问题讨论】：

在深入研究参与者位之前，您可能需要查看有关域模型中“集合验证”的文献。例如：codebetter.com/gregyoung/2010/08/12/…
嗯，链接好像挂了。
谢谢，这太棒了，帮助很大！
在actor中创建域模型实际上是错误的方法。当通常情况并非如此时，它会应用一刀切的范式。例如，您可能有一个用户参与者，它允许用户更改他们的个人资料信息，这是一个集群分片参与者，但另一个参与者提供有关用户活动的实时并发数据。在 SO 上 PM 我，也许我可以通过 Skype 电话帮助您。
@RobertSimmonsJr.不幸的是，我无法给您发送 PM。我可以通过其他方式联系您吗？

标签： scala akka domain-driven-design

【解决方案1】：

这可能是一个很长的答案，很抱歉没有 TLDR 版本。 :)

好的，那么您想“扮演”您的域模型吗？馊主意。领域模型不一定是参与者。有时它们是，但通常它们不是。为每个域模型部署一个参与者将是一种反模式，因为如果这样做，您只是将方法调用卸载到消息调用，但会丢失方法调用的所有单线程范例。您无法保证消息到达您的参与者的时间，并且基于 ASK 模式的编程是引入不可扩展系统的好方法，最终您有太多线程和太多未来并且无法继续前进，系统陷入困境和窒息.那么这对您的特定问题意味着什么？

首先，您必须停止将域模型视为单一事物，并且绝对停止使用 POJO 实体。我完全同意 Martin Fowler 在讨论贫血域模型时的观点。在构建良好的参与者系统中，通常会有三个领域模型。一种是持久模型，它具有为您的数据库建模的实体。第二个是不可变模型。这是参与者用来相互交流的模型。所有实体自下而上都是不可变的，所有集合都不可修改，所有对象只有 getter，所有构造函数都将集合复制到新的不可变集合。不可变模型意味着您的演员永远不必复制任何内容，他们只需传递对数据的引用。最后，您将拥有 API 模型，这通常是为客户端使用的 JSON 建模的实体集。 API 模型用于将后端与客户端代码更改隔离开来，反之亦然，它是系统之间的合同。

要创建您的参与者，请停止考虑您的持久模型以及您将使用它做什么，而是开始考虑用例。你的系统需要做什么？根据用例对参与者进行建模，这将改变参与者的实施及其部署策略。

例如，考虑一个向用户提供库存信息的服务器，包括单个供应商的产品的当前库存水平、用户评论等。用户敲定此信息，它会随着库存水平的变化而迅速变化。此信息可能存储在六个不同的表中。我们不会为每个表建模一个参与者，而是为这个用例服务一个单一的参与者。在这种情况下，这些信息会被重负载环境中的一大群人访问。所以我们最好创建一个actor来聚合所有这些数据并将actor复制到每个节点，每当数据发生变化时，我们都会通知所有节点上的所有复制者这些变化。这意味着获得概览的用户甚至不会接触数据库。他们点击演员，获取不可变模型，将其转换为 API 模型，然后返回数据。

另一方面，如果用户想要更改库存水平，我们需要确保两个用户不会同时执行此操作，但大型数据库事务会大大降低系统速度。因此，我们选择一个节点来保存该供应商的库存管理参与者，然后我们对参与者进行集群分片。任何请求都被路由到该参与者并按顺序处理。公司用户登录并记录收到 20 件新物品的收货。消息从他们命中的任何节点发送到持有该供应商的参与者的节点，供应商然后进行适当的数据库更改并广播所有复制的库存视图参与者拾取的更改以更改其数据。

现在这很简单，因为您必须处理丢失的消息（阅读有关为什么不需要可靠消息传递的文章）。然而，一旦你开始走这条路，你很快就会意识到，简单地让你的领域模型成为一个参与者系统是一种反模式，并且有更好的方法来做事。

反正那是我的 2 美分 :)

【讨论】：

现在开始变得更有意义了，还有何时使用 Actor 模型的用例。但是，您提供的用例表明存在许多用户想要检索的某种数据。因此，所有节点上的 Actor 都保持相同的状态，并在必要时进行更新。但是你将如何解决 f.e.从特定用户的银行账户中取款？您是否会向具有某种 ID 的 Actor 发送消息，从数据存储中检索特定的银行帐户并更新余额？或者你甚至不会使用 Actors 来做这样的事情？
绝对是演员。你说的这个案子是连环案。但它可以简单地建模。将一个事务写入包含所有内容的数据库。然后与参与者处理该交易。在这种情况下，您的 UserActor 可能是集群分片的（请参阅 akka 文档），银行也是如此。在我们的例子中，我们的 UserActor 还使用了 become() 来存储其他消息，直到这条消息完成。认为您需要银行的 ACID 命令式处理是一个错误的前提。你所需要的只是一个原子交易，你要么写钱包余额和交易，要么什么都不写。
很高兴知道我没有走错路。我尝试解决这个特定问题的方法（我只是为了更熟悉 Scala 和 Akka 而编写一个银行账户功能）是让 BankAccount Actor 以 AddTransaction 的形式接收消息。 Actor 最终将持久化一个被 Persistence Query 捕获的事件，该事件轮到他更新读取端（即余额）。考虑到我需要处理最终的一致性，这种方法有什么问题吗？
余额是交易汇总的一个方面。如果您可以保证交易已写入，您可以更改参与者不可变数据中的余额。向 AccountActor 发送带有金额的提款消息。它以原子方式写入交易和余额变化。这是一个简单的例子。资金在账户之间流动的例子更好。和以前一样，除了交易包含两个帐户 ID。事务由一个写入，另一个被通知更改并更新其状态。
购买是一个更有趣的例子，因为其中涉及库存参与者，但您所要做的就是记住，在现实生活中人们透支他们的帐户，然后一切都应该立即到位。现实生活的支持最终是一致的，而不是立即的

【解决方案2】：

一般设计

Actor 通常应该是业务逻辑的简单调度程序，并且包含尽可能少的功能。将 Actors 视为类似于 Future；当您希望在 Scala 中实现并发时，您无需扩展 Future 类，您只需围绕现有逻辑使用 Future 功能。

将您的 Actor 限制在最基本的责任范围内有几个优点：

无需构建 ActorSystems、探针、ActorRefs 等即可完成代码测试...
业务逻辑可以很容易地移植到其他异步库，例如期货和 akka 流。
使用普通的旧类和函数创建“适当的域模型”比使用 Actors 更容易。
在 Actors 中放置业务逻辑自然会强调更加面向对象的代码/系统设计，而不是功能性方法（我们选择 scala 是有原因的）。

业务逻辑（无 Akka）

在这里，我们将设置所有特定于域的逻辑，而不使用 any akka 相关的“东西”。

object BusinessLogicDomain {

  type FirstName = String
  type LastName = String 

  type Balance = Double

  val defaultBalance : Balance = 0.0

  case class Account(firstName : FirstName, 
                     lastName : LastName, 
                     balance : Balance = defaultBalance)

让我们将您的帐户目录建模为HashMap：

  type AccountDirectory = HashMap[LastName, Account]

  val emptyDirectory : AccountDirectory = HashMap.empty[LastName, Account]

我们现在可以创建一个函数来满足您对每个姓氏不同帐户的要求：

  val addAccount : (AccountDirectory, Account) => AccountDirectory =
    (accountDirectory, account) =>
      if(accountDirectory contains account.lastName)
        accountDirectory
      else 
        accountDirectory + (account.lastName -> account)

}//end object BusinessLogicDomain

存储库 (Akka)

既然未受污染的业务代码是完整且隔离的，我们可以在基础逻辑之上添加并发层。

我们可以使用 Actors 的become 功能来存储状态并响应请求：

import BusinessLogicDomain.{Account, AccountDirectory, emptyDirectory, addAccount}

case object QueryAccountDirectory

class RepoActor(accountDirectory : AccountDirectory = emptyDirectory) extends Actor {

  val statefulReceive : AccountDirectory => Receive = 
    currentDirectory => {
      case account : Account     => 
        context become statefulReceive(addAccount(currentDirectory, account))
      case QueryAccountDirectory => 
        sender ! currentDirectory
    }      

  override def receive : Receive = statefulReceive(accountDirectory)
}

【讨论】：