CQRS：是什么？ 为什么？ 怎么做？
作者：互联网
CQRS是推理特定域活动的有用模式。 但这肯定带有陡峭的学习曲线。 读取，写入，DDD，事件源，最终一致性，我们为什么要关心？
这是我在自己空间上的原始文章的交叉发布：CQRS：为什么？ 以及所有需要考虑的事情。 随意看看并查看我的其他文章。
Photo by Franck V. on Unsplash
用例
我正在研究一个新项目，该项目旨在替换会产生后果并在整个公司中引入耦合的大型（太大）整体。
为了适应不断变化的业务需求，缩短产品上市时间，改善团队之间的可追溯性和沟通，我们决定使用CQRS异步体系结构（但不提供事件源功能，稍后再介绍）。
长话短说，将所有人"转换"为新的思维方式并强加于人并非易事。 人们认为必须重新质疑，挑战和产生新的约束（例如最终的一致性，或者知道我们是否应该拥有某些数据）会受到质疑。 我曾经是这个家伙，总是说"你做不到"，提醒了为什么。
CQRS（与DDD结合使用）迫使我们施加强大的约束，并具有我们大多数人以前没有的限制（使用经典后端+ DB）。 这并不总是一件坏事：它揭示了一切。 我们对域本身了解很多，总是质疑域的分离和责任。 它使我们了解业务。
在这里，我将大致讨论CQRS及其与DDD和事件源的链接。 所有这些都始于一个用例，在该用例中，我们需要向其他服务公开最终不一致（也就是高度一致）的数据。 CQRS不一定要最终保持一致，但在我们的情况下，它是（因为异步且高度可用）。 我看到有些人皱着眉头，因为我们应该接受最终的一致性，这是生活（和分布式系统）的一部分。 同意 但..
我们为什么需要它？
CQRS（命令查询响应隔离）是非典型的年前，但是如今，我们可以找到很多有关它的内容，许多公司正在使用它并进行沟通。 我们甚至可以在不知道自己做的情况下进行CQRS！
模式很简单：
CQRS并不意味着实际上可以使用微服务，消息传递基础结构或事件，也不能进行DDD。 出于某种原因，它经常与这些技术一起使用。
一个写模型，多个读模型
CQRS的想法是使一个应用程序（广义上）能够与不同的模型一起工作：
· 它使用的一种内部模型：写模型，由Commands修改（稍后会详细介绍）
· 它和其他应用读取的一个或几个读取模型（我们不能让其他人读取写入模型）
读取模型可以由前端或API读取，没关系。
例如：如果我们有一个在某些资源上采用POST / PUT的API，将状态保存到PostgreSQL，并有一个将其同步到Elasticsearch上以在我们的网页上进行智能查找的过程，那么我们几乎在做CQRS（我们可能还会公开我们的 PostgreSQL ..）。
可扩展性
CQRS允许我们独立扩展系统：我们通常必须处理更多的读取而不是写入的操作，因此具有不同的可伸缩性。
我们希望在O（1）中访问读取：我们不希望它们花更多的时间来获取实体，因为它具有与其他实体的更多链接：我们想避免爆炸性的JOIN。 解决方案是在任何人请求之前，在数据更改时预先计算结果。 这样做还会在请求发生时使用更少的资源，从而减少延迟并使其在p999上可预测且稳定。 我们在争取空间。 稍后再详细介绍。
当我们实现Redis缓存以避免主数据库过载（进行写操作）时，这就是CQRS的精神。
业务需求
将写入模型与读取模型分开可以帮助我们分离域的复杂方面（谁需要什么，谁负责什么），并增加解决方案的灵活性。 我们可以更简单地适应不断变化的业务需求。
这是因为我们更多地考虑了责任：谁在变异数据？ 用例是什么？ 我们应该真正拥有这些数据吗？ 我们会采取行动吗？ 这不是其他应用程序的责任吗？ 谁只需要读取数据？ 应该是高度一致的吗？ 由于这种思维方式，CQRS通常与DDD相关联。
并发
从技术上讲，CQRS还可以通过显示并发和锁定（事务数据库）来简化它们。
当我们使用异步CQRS模式工作时，我们经常谈论数据的最终一致性，数据生命周期，数据所有权，业务需求以及许多关于建模的问题：实体的事务边界以及我们应始终拥有的不变式。 同样，这就是为什么CQRS通常是面向DDD的原因：必须非常仔细地定义数据形式的集合。 （稍后会详细介绍）
没有"读自己写"的语义
必须明确处理陈旧数据。 如果我们更改资源（通过发送命令）并立即读取相同的资源，我们将看不到更改。 异步CQRS不提供"读取自己的写入"语义。
前端可以通过进行乐观并发来模拟它：它可以嵌入一些知识，并假设它要求的突变会很好，因此它在获得真正的突变之前会显示它认为将是答案的东西。 如果有差异，它会适应。
当它是同步CQRS时，我们可以具有以下语义：我们将这两个模型在同一事务中写入不同的表中。 它始终是同步的。 CQRS很少是同步的，因为我们想使用不同的资源，数据库类型，使用消息传递基础结构来进行工作/扩展：我们很少能够使分布式事务跨资源（我们可以谈论XA或Sagas，但是……现在不行！）。
CQRS是一种模式，严格将处理命令（变异）的职责与处理无副作用查询/读取访问的职责区分开来。
这不是万能的解决方案
我们仅应在某些情况下考虑实施CQRS：
· 我们有很多不断变化的业务需求
· 公司不知道确切的去向
· 我们有可扩展性问题
· 我们与其他团队合作（即：其他受限环境）
· 多种服务竞争改变相同的资源
· 我们正在安排我们周围的其他服务
· 我们域中发生的事情会影响他们，反之亦然
· 我们的领域是面向写入的，我们不会读取您自己的数据，其他应用程序会
对于单独工作的简单API或明确定义范围的CQRS的开销可能是巨大且不必要的。
即使我们要实现CQRS架构，团队（开发人员和业务人员）也将需要绕过对变更的恐惧，遵循学习曲线，弯曲思路，然后适应过去的工作方式。
AKF规模立方
如果我们遵循Write + Reads逻辑，则意味着我们将以某种方式在系统中复制数据。 "相同"数据将出现在写入模型和读取模型下。
我遇到了一些真的很害怕重复数据的人，或者觉得这是一种反模式。 数据只有一个主机，它应该是控制谁有权访问数据的唯一主机，没有人可以在自己的数据库或消息传递基础结构中复制它。 您必须始终调用我的API！ 瓦德复古！
是的，有时可能需要始终调用API而不复制数据（例如GDPR下的用户数据）。 但是它很少是强制性的，它本身也可以成为反模式，因为它导致更多的复杂性，更多的依赖关系，不良的性能，服务中断，SLA减少。
我喜欢AKF规模多维数据集，以了解为什么组织中的重复是横向的：
简而言之，三轴：
· X：低级技术：我们应该复制，复制，缓存，负载均衡数据
· Y：组织：我们应该拥有独立的服务来处理不同的领域：它们负责自己的数据（按DDD表示）
· Z：高级分片：我们应该根据用例将相似的内容（按resource_id，按地理位置）细分为自己的基础结构
命令/书写：副作用
因此，导致对写模型进行写操作的事物称为命令。 这是设计我们始终使用的工具的通用术语：改变系统状态的工具（任何形式的更新）。
· 命令可以被同步或异步处理。
· 命令是否可以通过消息总线。
· 命令可以是简单的API调用。
· 如果不执行OOP，则Command可以是超类。
· 命令可以是简单的函数调用。
所有这些概念都与命令是正交的。 我们命令某些状态以某种方式改变。 命令是一种意图（不是事实），并且会导致副作用（对资源）。 它指向特定的目的地（不广播），并根据消费者域（共域）进行定义。
通常定义为VerbSomething：CreateOrder，ShipProduct，ComputePrice，GiveMoney。 稍后我们将看到定义为OrderCreated的事件的对称性（同样，事件不在CQRS范围内，但它们运行得很好）。
命令以行为为中心，而不是以数据为中心。 它是关于更改内容的意图，它并不映射到资源的格式（例如API中的DTO）。 它可以包含有助于处理意图的数据，仅此而已。
我们还将讨论基于任务的系统，而不是基于资源的系统。 写入部分不接受新资源或现有资源的补丁：它们接受任务（也称为命令）。 它可能已经被命名为TQRS :-)。
流
处理命令的流程始终相同：
· 如果是异步的，它将保存到消息传递基础结构中，并向呼叫者返回"确定"（不能做更多的事情）
· 在处理它时-同步还是异步-根据其形式（API，消息，函数调用），调用正确的命令处理程序（函数）
· 该处理程序必须确定是否有可能对其进行处理：
· ?如果必须采取行动（从数据库或使用事件源），它将检索当前状态
· ?它使用一些业务规则来知道是否可以在状态上授予或拒绝Command
· 如果被授予，它将在状态上应用命令（可以生成或不生成事件）
· ?如果未授予，则返回错误（同步或异步）
· 保存新状态
· 如果同步，它将返回OK到调用者-或最少的信息，例如ID，但不返回整个状态（即写模型）
· ?如果异步，则将消息提交到消息传递基础结构。
指挥部应遵循一劳永逸的原则：除"确定"外，我们不应期望指挥部有任何结果。 命令只是一个要求。 处理可以是同步的或异步的，我们不应该尽快获得结果。 结果将稍后来自读取的部分。
命令处理程序必须包含业务逻辑，以便能够立即拒绝或不拒绝命令。 例如，要处理
class OrderCommandHandler {
...
fun cancelOrder(cmd: CancelOrderCommand) {
val order=repo.fetch(cmd.orderId) // the only place where we can fetch the write model
if (order.inTransit()) error("Too late, we can't cancel an order in transit")
// ...
val newState=order.cancel()
repo.persist(newState) // OR persist only events, OR both atomically (Outbox Pattern)
}
关于命令还有很多要了解的内容，在这里我将不做介绍，因为这不是本文的重点。
· 命令处理应该是幂等的
· 命令不能使用读取端来获取数据
· 仅应出于技术原因重试命令，而不得出于业务原因重播命令：结果可能会有所不同（例如：此后更改了增值税）
· 可以将命令保存到命令总线中以进行进一步处理
命令是关于管理系统中的副作用的。
查询/读取/服务层：无副作用
如前所述，我们可以从相同的原始数据构建不同的读取模型。
当我们需要查询数据来回答不同的用例时，我们通常都会这样做。 我们在某些地方（PostgreSQL，MongoDB，Cassandra，S3…）有我们的真理源，在这里我们编写/更新我们的东西，但是我们想使用专门的数据库来不同地解释它们：
· 缓存数据
· 快速文本搜索
· 使用图形语言查询
· 处理时间序列数据
· 在一组维度内预先计算聚合
· 使用实时数据库（例如Google Cloud Firestore）向客户端发送实时更新
非规范化/合并
通常的做法是将我们的数据从关系数据库存储到Elasticsearch中，以享受快速搜索，自动完成，令牌生成器，令牌过滤器，评分，排名，稳定的延迟，而原始数据库通常无法获得这些功能。
如果我们在广告领域工作，并且想要有关广告商和发布者的实时仪表板，我们希望在时间序列数据库中预先汇总数据，因为我们关心按时间段和亚秒级响应进行汇总。 因此，我们将数据存储到Apache Druid或ClickHouse中，它们将在其中预聚合传入的数据（内部使用CQRS方法，啊！）。
我们正在对原始模型进行非规范化处理以适合其他模型：
· 以不同的方式显示数据（SQL到NoSQL）
· 到外部模型（公共），其他应用程序将使用（我们要隐藏实现）
· 不需要所有数据的"更轻便"模型
· 到"较重"的模型中，我们将添加许多我们不拥有的数据
最后一点很重要。 当我们更新读取数据库（由读取服务提供）时，通常会合并数据。 我们可以联接我们拥有的其他表或查询其他应用程序以获取我们不拥有/无法计算的数据（客户名称，产品说明，增值税等）。 我们不想在查询时执行此操作：我们想要可预测的稳定延迟，因此需要访问O（1）。
这种合并/转换是在我们需要更新数据库时（发生更改时）完成的，而不是在查询时进行的。 它的意思是：
· 我们可以更新永远不会读取的实体
· 我们可以更新将读取一百万次的实体
· 我喜欢的一个缩写是WORM：多次写入。 这是使用CQRS的最佳选择。
· 我们必须有一种方法来检测原始数据中的更改以触发读取数据库上的更新。 这是事件有用的地方。
此外，Reads数据库可能会令人恐惧，因为：
· 它可以包含多个域的数据
· 这意味着Reads Services应该由消费团队而非生产团队管理。 稍后再详细介绍。
· 可能会很大
· 可以在某种程度上从零开始销毁和重建它：假定其他服务具有幂等性。
· ?如果没有，重建可能会与以前有所不同。
· 我们可以升级我们的Writes服务，而不会同时关闭Reads Services（独立的生命周期）。
传统堆栈和新系统
在制作新产品时，通常希望将其集成到现有堆栈中（如果有的话）（例如，旧式堆栈）。
旧版不希望（或根本无法）适应我们的新模式或交流方式。 它只希望经典API重用它了解的现有格式来获取数据。 这是专用阅读服务（"传统阅读服务"）的理想用例。 它可能需要不同的服务集来构建遗留视图，并需要不同的业务规则来匹配遗留模型。
传统模型倾向于具有更多数据（由于采用整体方法），可以是跨域的（现代堆栈由于DDD方法而被拆分），并且可能继承了更旧的堆栈约束。 大规模发布是不可行的，因此我们必须确保服务的连续性和重叠性。
没有副作用
具有多个数据库强加了一个约束："读取服务"必须仅处理读取，而不处理写入。 它们不是真理的源头。
它们和Writes数据库之间没有同步。 读取数据库不能在写入数据库上产生副作用。 由于非规范化，甚至不可能找回原始记录（如果我们不保留PK之类的所有必要信息，因为我们不需要它们）
当我们查询API（GET或某些GraphQL查询）时：我们不希望对数据进行突变，这是相同的。
最终一致性
如前所述，使用Reads Service的用户可以处理过时的数据。
当我们发送要处理的命令时，它的处理可以异步完成（无结果，即焚即忘），并且读取数据库的更新将是异步的（除非它是同一数据库，但很少见？）。
例：
· T0：我们发送有关订单1234（v1）的AddProductToOrderCommand。
· T1：Writes Service更新1234号订单（v2）的项目。
· T2：通知Reads Service，并查询产品服务以将其视图与名称和描述合并（进行中）。
· T3：还收到通知的外部电子邮件服务，通过查询Reads Service（v1）来请求订单的详细信息。
· T4：产品服务响应读取服务，该服务最终可以更新其读取数据库（v2）。
在T3，尽管先前的更新发生在T2之前（绝对时间），但Reads Service发送了Order（v1）的旧版本，因为它不是最新的。 我们不知道系统收敛需要多长时间。 这就是为什么CQRS没有"读自己写"的语义的原因，也是为什么当外部系统与我们交谈时我们必须始终考虑最终的一致性。
我们刚介绍的版本在该系统中必须具有。 这表示一个逻辑时间，一个单调的计数器，可以帮助我们了解流程的进展情况并制定决策。 它是Writes模型的一部分，并传播到整个系统（如果有事件，则读取模型…）。
版本不是我们在此类系统中可以找到的唯一人工制品。 我们还可以使用向量时钟和CRDTs结构来处理事件排序因果关系。 这些通常是外部系统不想处理的事情，这是我们内部的混乱，因此需要一个独特的读取模型来消除噪声。
有一些技术可以帮助外部系统获取他们期望的版本，稍后再进行介绍。
如果我们希望在读写数据库之间保持高度一致性，则需要对它们进行原子更新（由于所使用的异构系统，这种情况很少出现）。 即使这样，如果这样做，我们将失去依靠不同系统提供的可用性（CAP定理，我们可能是处于故障状态的CP或AP，而不是同时处于一致状态和可用状态）。
通过将两个系统清楚地分开，读取服务独立于写入服务工作：不同的生命周期，不同的部署，不同的约束，不同的SLA。
一套不同的功能
读取服务可以提供比写入服务更多的功能。 双方的参与者和用例都不相同。 这些功能可以是：
· 认证方式
· 快取
· 加密
· 限速
· 保留期有限
· 弹性缩放
· 复制和LB
· 分片
· SSE或Websocket用于实时更新（推送）
当我们具有用于读写的相同数据库时，我们会将不同的关注点混在一起。
· 假设我们有一个电子商务网站，它使用一项阅读服务向客户显示购物车和订单。 通信量大（面向客户），读取服务被复制或分片，依靠缓存，进行网络套接字。
· 另一方面，我们公司的员工拥有自己的管理应用程序，他们需要查看和更新订单以准备订单，打包订单，更新交货状态等。他们不需要花哨的功能，流量 不重，过程很慢。 他们只是需要99.999％的正常运行时间，因为这是他们工作的一部分。
两种Reads Service都有不同的SLA（一个可以"降级"，而另一个不可以），功能和模型（admin Reads Service将提供更多内部详细信息，对客户Reads Service而言是隐藏的）。
如何建立读服务
作为开发人员，我们喜欢测试新技术，新数据库，新系统。 拆分我们的写作方式和阅读方式的奇妙想法为我们提供了实现这种爱所需要的东西。
创建新的Reads Service（具有其独特的Reads Database）并在现有流程中进行并行测试，没有比这容易的了。 如果我们处理事件，则只需订阅事件并构建新模型。 如果我们没有任何事件，那么我们可能有一个来自数据库的同步程序，我们可以更改或复制它以写入新系统。
而且，我们可以有多种语言。 每个阅读服务都可以用任何语言编写，以更准确地回答读者使用的用例和技术。
这是构建此类服务的技术列表：
双重写入
· 在代码中，当我们使用（可爱的）ORM（或不使用）写入数据库X时，我们添加代码或抽象以也写入Y
· 我们必须使用交易或适当的交易来确保它们之间的原子性/一致性
数据库同步
· 每N分钟复制和转换数据的一批
· 一种"几乎实时"的后台服务，每N分钟轮询一次更改
更改数据捕获（CDC）
· 它将数据库更改导出到事件流中
· 它依靠复制日志充当数据库发出的事件； 然后我们可以对它们进行一些流处理
· 这是Kafka Connect与Debezium（PostgreSQL，MySQL）的绝佳用法之一
大事记
· 我们订阅现有事件（在原始状态更改时发布），并根据它们建立我们的读取状态（我们可以从过去重放它们）到任何数据库
· 我们还可以使用Kafka Streams及其交互式查询来创建分布式自动分片数据库。
隐藏Lambda架构
· 与批处理视图合并的实时视图
最好的方法是依靠pub/sub系统的事件。 从本质上讲，它们已经使发布者与消费者脱钩。 这是CQRS通常与事件相关联的另一个原因。
DDD在哪里？
到目前为止，我们几乎没有谈论过DDD，它经常与CQRS关联（但不是强制性的）。 为什么？
我们说我们有一些命令，这些命令的处理程序会检查业务需求，如果可以，请更新"状态"并将其保存到Writes数据库中。
在DDD中，此状态包含在我们所谓的聚合中。 它是一棵实体树（具有实体根），该树是独立的，自治的，并且从外部和业务的角度来看始终保持一致。
集合必须确保其实体内的交易边界：从外部看，我们永远看不到集合"半形成"，"半变换"，"半有效"。 我们无法直接访问子实体：所有内容都只能引用集合（以控制一致性）。 始终遵守业务不变性规则（例如："如果尚未付款，则订单不可转让"（好的，这要取决于我们的想法了！））。
在DDD中，实体具有特殊含义。 这是一个由其唯一标识而不是由其属性（例如User（userId））定义的对象。 Color（r，g，b）不是实体，其属性定义了它，它是一个Value Object，它是不可变的，因此可以共享。
聚合通常是OOP中具有所有与业务相关的代码的不可变且具有副作用的免费类（与DTO相反，DTO通常是贫乏的，即：没有业务逻辑）。 它不依赖于任何序列化框架（没有糟糕的JSON注释，没有Avro生成的类..），它没有任何注释等。
这是企业应该理解的简单，可重复使用的代码，因为它使用的是他们的话：这是无所不在的语言。 使用相同的语言可以减少混淆和误解，因为我们不会在"技术"和"业务"方面进行隐式（或宽松地）翻译。
// our aggregate with a private constructor
case class Order private(id: OrderId, items: List[LineItem], state: OrderStatus) {
def addItem(item: LineItem): Order=?
def removeItem(item: LineItem): Order=?
def canBeCancelled(): Boolean=?
def startProcessing(): Order=?
def addCoupon(): Order=?
def cancel(): Order=?
// Optional: events generated by the previous methods to be persisted separately
private val events: List[OrderEvents]=List()
}
object Order {
// the public constructor ("factory method")
def apply(): Order=Order(randomOrderId(), List(), OrderStatus.CREATED)
}
聚合由存储库（具有自己的持久化方法，已隐藏）检索并持久化。 聚合不代表基础数据库或序列化格式的确切映射。 它是面向业务的，而不是面向技术的。
class OrderRepository {
def fetch(id: OrderId): IO[Option[Order]]=?
def save(order: Order): IO[Unit]=?
}
回到命令，我们在处理聚合时真正发生了什么：
· 代码将其定向到命令处理程序以对其进行处理
· CH从存储库中获取聚合（在命令中引用）
· Command命令可以与现有资源有关，也可以请求创建一个资源，例如：CreateOrderCommand
· CH会检查一些业务规则，以查看命令是否符合业务规则
· 是的，它在聚合上调用必要的函数。
· these这些函数中的每一个都返回聚合的新状态。
· 聚合的结束状态将保留在Writes数据库中（或将事件存储到事件存储中）。
在复杂的域中，DDD使我们能够以商业方式构造命令和事件，每个理解该域的人都可以理解。 DDD有助于发现界限，限制责任并使系统的不同部分可维护，因为它们具有"合理性"。 这样的系统并非仅凭开发人员的有机发展。
查找命令和聚合以及更多
查找集合及其命令（以及事件等）的一种流行做法是与开发人员，业务人员和专家进行事件风暴研讨会。
通过查找我们的域必须处理的所有可能事件，我们可以将它们重新分组并形成包含相关事件的聚合。 由此，我们形成了具有凝聚力的子域（事物属于一起），我们形成了实体，集合并同意了泛在语言。
另一种技术是域故事讲述。 我们想到了一个用户场景。 我们绘制它来显示流程中的人工制品和人员（从何处，何人，往何处去，由谁，何处，谁做出反应，谁发送东西等进行验证）。 有4个项目：演员，工作项目（文档，交流），活动（动词），（评论）。
如果您感到好奇，也可以检查业务模型画布。
大事记
使用CQRS广播事件不是强制性的。 这是一种自然的工作方式。 更新聚合时会发出事件。 将它们广播给"谁想听"（其他人和自己）。 使系统具有反应性和独立性非常有用。
这是事件列表：
case class OrderCreated(orderId: OrderId, items: List[LineItem], customer: CustomerId)
case class OrderLineAdded(orderId: OrderId, item: LineItem)
case class OrderCancelled(orderId: OrderId, reason: Option[String])
...
现在，我们可以看看依赖于CQRS，聚合，事件的更完整的图片：
Raison d'être
当我们发生事件时，他们通常会成为一等公民。 这意味着一切都围绕它们构建：业务逻辑，依赖关系（"在执行Z之前我们需要X和Y"）。
作为命令，事件是一个通用术语，它不定义其实现，而是更多其行为和起源。
事件是过去的事实。 它本质上是一成不变的（我们不能改变过去的权利吗？）。 与具有固定目的地的命令不同，它们可以包含其起源，是有目的的，而事件则相反：事件只是以一种"忘却"的方式向世界广播，是无目的的，并且可以是匿名的。 因为我们不知道谁在听，所以我们无法对谁需要它进行硬编码（我们仅依靠传输事件的技术总线）：这减少了系统之间的耦合。 我们可以创建新的监听系统，而无需发出系统来了解和关心。
这就是像Kafka这样的发布/订阅系统有用的地方，因为它们是这种中间总线，可以保持事件，处理可以动态添加或删除的分发给消费者的东西等等。
重要的是，事件在生产者域中定义。 在使用者域中定义了一个命令（或更通用的"消息"）（我们说目标的语言，给它一个"命令"）。
投影事件
事件可以由任何事物创建。 在执行CQRS和DDD时，它们主要由聚合创建。 更改汇总后，它将发出一个或多个与更改相对应的事件。 事件还可以由诸如调度程序之类的外部事物产生，这取决于时间："在小时结束时发送此事件X"。
在这里，我们可以选择是否进行事件采购。
· 我们可以决定独立于聚合事件的生成来更新其状态：
def increment(state: State): (State, Event)={
val newState=state + 1
(newState, NumberWasIncremented(1))
}
val (newState, event)=increment(5)
· 我们可以决定通过"播放"当前状态的事件来创建聚合的新状态：
def increment(): Event=NumberWasIncremented(1)
def apply(state: State, e: Event): State=e match {
case NumberWasIncremented(x)=> state + x
}
val newState=apply(5, increment())
Event-Sourcing方式可能看起来过度设计。 我们还需要一个功能应用和模式匹配。 我们已经将事件创建的逻辑与事件应用程序分开了。 这样做比较容易。 申请活动时，我们不必知道活动的来源。
一个事件是轻量级的，它仅与变化有关。 通过重播其生命中的所有事件，我们可以从头开始创建聚合。 我们对发生的事情有更多的了解，我们不仅保留最新状态，也不保留更改之间的快照。 我们自己保留更改。 状态是一个汇总，是事件的投影。 我们经常具有一次重播所有内容的功能，有时称为计算器，reduce，replayer：
def applyAll(initialState: State, events: List[Event]): State={
events.foldLeft(initialState) { case (s, e)=> apply(s, e) }
}
进行事件采购的优势之一就是我们还可以以不同的方式重放事件：这将形成不同的状态。 当我们的模型经常变化时，这是完美的：事件没有变化，但我们对它们的解释有所不同。 当我们的模型因业务模式不断变化而动态变化时，这很有用。 诸如Kafka之类的技术使消费者可以重播过去的事件（以重建其状态/聚合）并赶上当前的事件（通过重置偏移量）。
与CQRS和DDD一起使用时，我们倾向于发出事件，因为它们代表了业务现实：业务中发生了某些事情！ 过去不会改变，因为我们想添加新功能。 我们只能解释过去。
我们可能有现有的用例来处理事件，但是将来我们可能会发现新的用例。 这就是为什么我们不想丢弃任何东西，而我们希望存储所有事件的原因：以后以未知的新方式（出于可追溯性和更多原因）处理它们。 事件是黄金。 不要失去他们。
事件并不意味着事件源
不是因为我们要处理事件，而是必须进行事件源。 如果我们不从事件中构建汇总：我们不是从事件中进行采购，因此不是事件中的采购。 QED。
（或没有）事件来源的逻辑隐藏在AggregateRepository的实现中。 这只是获取集合的一种特定方式。
当我们进行事件采购时，我们不是直接从数据库中获取最新状态，而是要求事件存储获取有关特定aggregationId的所有事件（因此，对于事件存储，Kafka不是一个好的选择，但是Kafka Streams可能是 ，以及"互动式查询"），我们会重播它们以建立最新状态。 这也意味着我们必须将所有事件保留在历史记录中以进行事件搜索。
有一些策略可以每X个事件对状态进行快照，从而避免在大量更新聚合时每次重播10000个事件。 但这是另一篇文章。
事件源是CQRS的正交概念。 我们可以一无所有。
智能端点和哑管
马丁·福勒（Martin Fowler）介绍了"智能端点，哑管"的概念。 简而言之：不要在传输机制中加入业务逻辑。 我们必须控制我们的业务逻辑。 我们不希望某些管道成为瓶颈（因为ESB成为大公司）。 运输机制应该保持愚蠢：它在技术方面就在那里，仅此而已。
这让我想起了ReactJS，我们在这里谈论：
-愚蠢的组件：纯净，呈现一段UI，无逻辑（按钮，日历，菜单）-智能组件：处理数据，具有状态和一些业务逻辑。
智能组件依赖于哑组件。 我们可以更改智能组件的实现方式，只需更改其使用的哑巴组件即可：我们不会更改其逻辑。
发布或使用的系统很聪明：它们处理映射，域转换等（使用反腐败层，当不同的有界上下文相互交谈时使用DDD概念）。 它具有更高的可扩展性，通常更易于维护，因为每个人都有自己的逻辑并提出要求。 自己动手
让某些州居住在ESB中也很常见，因为我们想根据过去或某些动态情况来更改行为。 智能终端将创建并维护自己的私有状态。 这不是"管道"的作用。
格式和语义
我们知道为什么发出事件很好，但是应该遵循什么格式？ 有标准吗？ 我们应该存入他们什么？ 那元数据呢？ 如果我不进行事件采购但想生成事件该怎么办？有什么区别？ 事件是一个简单的词，但具有巨大的语义。
没有什么是"官方的"，但是我们可以依靠或从中获得启发：
· 这提供了与JSON-LD兼容的基于JSON的语法。
· 语境
· 一种活动
· 演员
· 一个东西
· 潜在目标
· CloudEvents更有用，更完整，它提供了不同语言，不同序列化格式（Json，Avro，Protobuf等）的规范和SDK，并且可能最终落入CNCF（云原生计算基金会）。 它是面向无服务器的，但是事件是事件吧？
{
// CloudEvents metadata data goes here
"eventType": "order.line.added",
"eventID": "C1234-1234-1234",
"eventTime": "2021-08-08T14:48:09.769Z",
"eventTypeVersion": "1.0",
"source": "/orders",
"extensions": {},
"contentType": "application/json",
"cloudEventsVersion": "0.1",
// Custom event data goes here
"data": {
"orderId": "abc",
"item": { "reference": "prod123", "quantity": 12 }
}
}
关于语义，Mathias Verraes（DDD专家）在列出并解释事件可以拥有的不同语义方面做得很好。 以下是简短摘录：
· 摘要事件：将多个事件合并为一个（将多个项目添加到项目收藏夹中）
· 时间流逝事件：调度程序发出事件而不是命令DayHasPassed
· 隔离事件：将事件从一个域转换为另一个域
· 节流事件：每个时间单位仅发出最新事件
· 变更检测事件：仅当事件流中的值更改时，才产生新事件
· 胖事件：向事件添加冗余信息以减少使用者的复杂性
· …
注意最后一个事件："胖事件"。 我们将很快使用此策略将聚合的完整状态嵌入事件本身。
内部VS外部事件
有关语义和DDD的更多信息：事件可以是内部的或外部的。
· 内部事件是由我们的域（受限制的上下文）产生和使用的，它是私有的。
· 外部事件将由我们无法控制的其他域（公开）使用。
· ?我们自己的应用程序将使用来自其他方的外部事件。
· ?外部事件具有使人们知道如何使用它们的架构。 它被共享到元数据服务或某些注册表（或Excel ..）中。
两种类型都有不同的目标，因此约束和用法也不同。
内部事件倾向于规范化，并且主要包含引用（ID）。 它与内部模型一起使用。 我们不想添加无用的信息或重复我们系统其他部分已经拥有的信息。 参考文献越多，对未来的发展越好。 我们将能够重构模型而无需更改事件。
与外部模型一起使用的外部事件是供公众使用的。 它们经常被非规范化（包含名称，地址，没有版本，更简单），以避免消费者理解依赖关系。
将内部事件暴露给外部会导致复杂的体系结构，困难的开发，缺乏可见性并模糊域边界。 如果我们这样做，却没有注意到，就将内部事件转换为外部事件。 因此，这也将我们的内部模型转换为外部模型。 我们正在与其他服务之间引入强大的耦合。
这就是本文的重点：为什么以及如何保护我们免受此侵害。
从内部到外部
一种解决方案是进行一些流处理，以转换和合并事件（通过调用其他服务）以使事件成为外部事件。 既包含胖事件又包含隔离事件层。 这是一个从私有（有点复杂，看到明显的区别）到公共转换的示例：
事件进行状态转移
有时，出于性能和简化的目的，我们希望将聚合的状态与更新时生成的自身事件一起包括在内。
确切知道要放入事件中的内容可能很困难。 如果我们忘记了某些东西，或者对于消费者（内部或外部）来说不够用，我们可以引入网络/请求开销，并使他们的工作复杂化。
如果事件是普通事件/裸体事件（只是发生了什么变化，没有最新状态），则他们可能需要更多信息，并且需要根据过去的所有事件建立自己的状态（即：充当读取服务；它可以是KTable （如果使用Kafka Streams）或请求现有的Reads Service进行工作。
这给消费者带来了一些问题：
· 他们不想做有状态的并复制一个已经存在的状态
· 如果他们只关心一种类型的事件（例如OrderShipped发送电子邮件），则他们不希望监听所有事件来建立状态（OrderCreated，OrderItem Transactions，OrderShipped等）来构建状态，他们需要 订单，但OrderShipped只能具有orderId）
· 他们想查询一些读取服务以获取缺少的内容
· ?这会引入最终一致性：他们可能无法接受。
因此，一种解决方案是将状态嵌入事件本身。 这是事件进行的状态转移。
想象3个简单事件：
{ type: "OrderCreated", orderId: 1337, customer: "john" }
{ type: "OrderLineAdded", orderId: 1337, item: { ref: "prodA", qty: 2 } }
{ type: "OrderShipped", orderId: 1337 }
如果我们的电子邮件服务仅侦听OrderShipped事件，则其中没有商品，因此需要查询可能尚未处理OrderLineAdded（滞后，网络问题）的读取服务。 发送的电子邮件中不会包含任何内容，或更糟的是，阅读服务甚至可能不知道orderId 1337。
一种解决方案是在事件上添加一个版本（每个事件都会增加其表示的聚合的版本），然后我们可以在Reads Service中请求特定版本：
{ type: "OrderCreated", orderId: 1337, v: 1, customer: "john" }
{ type: "OrderLineAdded", orderId: 1337, v: 2, item: { ref: "prodA", qty: 2 } }
{ type: "OrderShipped", orderId: 1337, v: 3 }
GET /reads/orders/1337?v=3
然后，读取服务可能会阻塞，直到它获得预期的版本或超时（如果从未发生过）。
避免依赖关系，网络调用和阻止请求的另一种解决方案是使用此事件承载状态转移策略（我们将状态放入事件中）：
{ type: "OrderCreated", orderId: 1337, customer: "john",
state: { id: 1337, customer: "john", items: [], state: "CREATED" }
}
{ type: "OrderLineAdded", orderId: 1337, item: { ref: "prodA", qty: 2 },
state: { id: 1337, customer: "john", items: { ref: "prodA", qty: 2 } }, state: "CREATED" }
{ type: "OrderShipped", orderId: 1337,
state: { id: 1337, customer: "john", items: { ref: "prodA", qty: 2 }, state: "SHIPPED" } }
}
对于消费者而言，这是一项毫无头脑的工作。 他们只需要阅读"状态"字段，该字段就始终使用相同的模型（这里是"订单写入模型"）来完成工作。
开销现在在消息传递基础结构上（消息更大，因为事件现在包含整个状态）。
保护写模型
上面的示例仅适用于内部使用，不适用于外部服务（例如此处的市场营销服务），因为我们不想公开我们的写入模型。
私人_wm：WriteModel;
Write模型是Writes端的模型。 它是我们域负责的汇总模型。
如果我们碰巧像以前那样使用事件状态策略来公开它，则第三方服务可以将其逻辑基于状态，这就是我们的Write模型。 因此，在维持消费者兼容性（最大程度上）的同时，我们将无法发展自己的模型。
如果我们正在执行CQRS，则不需要这样做。 如果您还记得的话，当我们有不断变化的业务需求时，CQRS会很有用，因此变化会发生很多。 我们不想通知消费者我们已经更改了格式，不赞成使用当前主题，进行了重大更改等等。 我们想隔离我们的内部模式和我们在事件中公开的内容。
如果我们需要公开它怎么办？
商界人士很少会考虑最终一致性，这对他们来说毫无意义（"这是技术性的，我不在乎"）。 他们希望使用反应式架构，微服务，可在全球范围内扩展，具有可用的弹性一致服务（使用最新状态）。 不幸的是，在这些约束条件下，我们通常最终会遇到一个最终一致性系统（AP）。
从技术上讲，我们可以快速考虑在Write Service上添加/ get路由并完成此操作。 "为什么要麻烦读服务使数据最终保持一致，却没有任何意义"。 它将起作用，但是由于本文列举的所有原因以及CQRS体系结构中的反模式，这将是一个严重的错误。 我们决定使用CQRS，我们必须尊重它的工作方式。
· 除了写模型的系统之外，没人应该知道写模型。
· 由于可伸缩性原因，我们选择CQRS
· 如果我们向读者展示我们的Write Model，那么它就不是任何人的Write Model，而是Write / Read模型，没有区别了，我们无法以不同的方式缩放它们。
· 我们无法公开发布包含写入模型（事件承载状态）的事件
如果另一个服务必须绝对拥有我们的最新书面数据，我们应该问自己：
· 为什么，它的业务案例是什么？
· 为什么它不能处理最终一致性？ 有事实和数字支持的真正原因吗？
· 为什么我们的事件未涵盖其用例？
· 我们应该发送命令而不是被动地发送命令吗？
· 我们是否有错误的域分隔？ （应该在使用Commands而不是在外面）
ACL抢救
解决我们的Write模型的私密性（而不是解决可伸缩性问题）的最后一种解决方案是在Writes Service中编写一个ACL（反腐败层）。 我们可以将写入模型转换为读取模型。
· 即时运行，但将我们的Writes Service与读取结合起来：
· 或将其保存在数据库中，以避免即时转换（仍然耦合）：
在后一种情况下，必须在同一事务中以原子方式保存写模型（或事件存储中的事件）和此读模型（在发生故障时不会不一致）。 另外，我们也无法调用外部API：如果它们失败了，我们将无法完成转换，因此不会提交事务，因此所有命令都将失败。 SLA=最小值（取决于SLA）。 这将破坏CQRS的目的。
结论对企业
当我们习惯使用CQRS时，它是推理整个系统的好模式。 肯定存在陡峭的学习曲线，以了解分离和约束。 而不是全局考虑，我们必须本地考虑（命令，事件）。 很难预测总体行为。 这些功能分为几部分，我们可以有不同的参与者和不同的时间表来完成它们（更多情况下是失败的）。 最终的一致性还可以增加我们以前采用整体式设计所没有的问题。 我们对可扩展性和灵活性（CQRS）进行控制贸易。
通常，CQRS会事先结合良好的DDD思想，以避免混淆不同的域（或我们不应该使用的反向拆分域），或者创建毕竟不适合业务的奇怪聚合。 写和读模型之间的区别有很大帮助。
我经常问："为什么在写入模型中有此属性，我们对此有业务规则吗？"。 如果答案是"否，但这是针对下游内部服务的。"那么我知道它不属于这里。 Write Model上的每个属性都应在至少一个CommandHandler中具有一条业务规则。
关于Read模型，完全是YOLO，FFA，因为这无关紧要，没有影响。 它的目的是根据前端或消费者的需求进行大量改进。 我只想保护写模型。
CQRS还可以与事件一起使用。 他们是关于生意的。 它们提供了可追溯性，并在其他方面解耦了系统。 Event Sourcing方法本质上非常有用且非常安全，但是它附带了自己的约束（性能，快照，存储..）。 处理事件意味着我们拥有一个事件驱动系统。 这样，一项功能可能需要通过消息总线进行20跳才能完成（一系列边界事件处理程序/命令处理程序）。 如果没有良好的性能调整和良好的系统可观察性（日志/指标/跟踪，以及对事物之间如何关联的良好理解），它的性能可能会很差。 即使可以扩展，也不会有效。
..高级内容！
此外，必须注意一些技术上的预防措施，例如幂等和一次处理（由于瞬态网络问题，我们不想对一个事件进行两次处理）。 给定聚合的事件必须按顺序处理（您不希望在OrderCreated之前添加OrderItemAdded）。 事件还意味着我们将讨论事件流处理。
此外，当在处理事件的中间发生错误（反序列化问题，意外值，网络错误）时，我们的处理可能会停留在处理大型功能的过程中（拆分为多个事件）：我们该怎么办？ 结果是什么？ 我们如何发出错误，谁来消耗它？ 我们如何联系用户以告诉他（命令已经消失了一段时间）？ 我们如何确保状态一致？ 我们甚至没有谈论Sagas和错误管理，因为这是一个完整的故事。
我们还必须考虑对聚合和事件进行版本控制，必要时管理因果关系（向量时钟或矩阵时钟）。
所有这些问题不是由CQRS本身引起的，而是因为我们正在使用分布式系统。 如果您的公司希望从整体架构过渡到CQRS架构，则需要考虑所有方面：这是您需要坚强的肩膀（..要哭泣）的地方。
这是我在自己空间上的原始文章的交叉发布：CQRS：为什么？ 以及所有需要考虑的事情。 随意看看并查看我的其他文章。
谢谢阅读！
标签：怎么,为什么,读取,模型,数据库,事件,CQRS,我们 来源： https://blog.csdn.net/iqifenxia/article/details/122269309