基于Actor模型的CQRS、ES解决方案分享

开场白

你们晚上好，我是郑承良，跟你们分享的话题是《基于Actor模型的CQRS/ES解决方案分享》，最近一段时间我一直是这个话题的学习者、追随者，这个话题目前生产环境落地的资料少一些，分享的内容中有一些我我的的思考和理解，若是分享的内容有误、有疑问欢迎你们提出，但愿经过分享这种沟通方式你们相互促进，共同进步。

引言

1. 话题由三部分组成：

• Actor模型&Orleans：在编程的层面，从细粒度-由下向上的角度介绍Actor模型；
• CQRS/ES：在框架的层面，从粗粒度-由上向下的角度介绍Actor模型，说明Orleans技术在架构方面的价值；
• Service Fabric：从架构部署的角度将上述方案落地上线。

2. 群里的小伙伴技术栈可能可能是Java和Go体系，分享的话题主要是C#技术栈，没有语言纷争，彼此相互学习。好比：Scala中，Actor模型框架有akka，CQRS/ES模式与编程语言无关，Service Fabric与K8S是同类平台，能够相互替代，我本身也在学习K8S。

Actor模型&Orleans（细粒度）

1. 共享内存模型

多核处理器出现后，你们经常使用的并发编程模型是共享内存模型。

这种编程模型的使用带来了许多痛点，好比：

• 编程：多线程、锁、并发集合、异步、设计模式（队列、约定顺序、权重）、编译
• 无力：单系统的无力性：①地理分布型②容错型
• 性能：锁，性能会下降
• 测试：
  • 从坑里爬出来不难，难的是咱们不知道本身是否是在坑里（开发调试的时候没有热点多是正常的）
  • 遇到bug难以重现。有些问题特别是系统规模大了，可能运行几个月才能重现问题
• 维护：
  • 咱们要保证全部对象的同步都是正确的、顺序的获取多个锁。
  • 12个月后换了另外10个程序员仍然按照这个规则维护代码。

简单总结：

• 并发问题确实存在
• 共享内存模型正确使用掌握的知识量多
• 加锁效率就低
• 存在许多不肯定性

2. Actor模型

Actor模型是一个概念模型，用于处理并发计算。Actor由3部分组成：状态（State）+行为（Behavior）+邮箱（Mailbox），State是指actor对象的变量信息，存在于actor之中，actor之间不共享内存数据，actor只会在接收到消息后，调用本身的方法改变本身的state，从而避免并发条件下的死锁等问题；Behavior是指actor的计算行为逻辑；邮箱创建actor之间的联系，一个actor发送消息后，接收消息的actor将消息放入邮箱中等待处理，邮箱内部经过队列实现，消息传递经过异步方式进行。

Actor是分布式存在的内存状态及单线程计算单元，一个Id对应的Actor只会在集群种存在一个（有状态的 Actor在集群中一个Id只会存在一个实例，无状态的可配置为根据流量存在多个）,使用者只须要经过Id就能随时访问不须要关注该Actor在集群的什么位置。单线程计算单元保证了消息的顺序到达,不存在Actor内部状态竞用问题。

举个例子：

多个玩家合做在打Boss，每一个玩家都是一个单独的线程，可是Boss的血量须要在多个玩家之间同步。同时这个Boss在多个服务器中都存在，所以每一个服务器都有多个玩家会同时打这个服务器里面的Boss。

若是多线程并发请求，默认状况下它只会并发处理。这种状况下可能形成数据冲突。可是Actor是单线程模型，意味着即便多线程来经过Actor ID调用同一个Actor，任何函数调用都是只容许一个线程进行操做。而且同时只能有一个线程在使用一个Actor实例。

3. Actor模型：Orleans

Actor模型这么好，怎么实现？

能够经过特定的Actor工具或直接使用编程语言实现Actor模型，Erlang语言含有Actor元素，Scala能够经过Akka框架实现Actor编程。C#语言中有两类比较流行，Akka.NET框架和Orleans框架。此次分享内容使用了Orleans框架。

特色：

Erlang和Akka的Actor平台仍然使开发人员负担许多分布式系统的复杂性：关键的挑战是开发管理Actor生命周期的代码，处理分布式竞争、处理故障和恢复Actor以及分布式资源管理等等都很复杂。Orleans简化了许多复杂性。

优势：

• 下降开发、测试、维护的难度
• 特殊场景下锁依旧会用到，但频率大大下降，业务代码里甚至不会用到锁
• 关注并发时，只须要关注多个actor之间的消息流
• 方便测试
• 容错
• 分布式内存

缺点：

• 也会出现死锁（调用顺序缘由）
• 多个actor不共享状态，经过消息传递，每次调用都是一次网络请求，不太适合实施细粒度的并行
• 编程思惟须要转变

---

第一小节总结：上面内容由下往上，从代码层面细粒度层面表达了采用Actor模型的好处或缘由。

---

CQRS/ES（架构层面）

1. 从1000万用户并发修改用户资料的假设场景开始

1. 每次修改操做耗时200ms，每秒5个操做
2. MySQL链接数在5K，分10个库
3. 5 *5k *10=25万TPS
4. 1000万/25万=40s

在秒杀场景中，因为对乐观锁/悲观锁的使用，推测系统响应时间更复杂。

2. 使用Actor解决高并发的性能问题

1000万用户，一个用户一个Actor，1000万个内存对象。

200万件SKU，一件SKU一个Actor，200万个内存对象。

• 平均一个SKU承担1000万/200万=5个请求
• 1000万对数据库的读写压力变成了200万
• 1000万的读写是同步的，200万的数据库压力是异步的
• 异步落盘时能够采用批量操做

总结：

因为1000万+用户的请求根据购物意愿分散到200万个商品SKU上： 每一个内存领域对象都强制串行执行用户请求，避免了竞争争抢； 内存领域对象上扣库存操做处理时间极快，基本没可能出现请求阻塞状况；

从架构层面完全解决高并发争抢的性能问题。 理论模型，TPS>100万+……

3. EventSourcing：内存对象高可用保障

Actor是分布式存在的内存状态及单线程计算单元，采用EventSourcing只记录状态变化引起的事件，事件落盘时只有Add操做，上述设计中很依赖Actor中State，事件溯源提升性能的同时，能够用来保证内存数据的高可用。

4. CQRS

上面1000万并发场景的内容来自网友分享的PPT，与咱们实际项目思路一致，就拿来与你们分享这个过程，下图是咱们交易所项目中的架构图：

开源版本架构图：

（ 开源项目github：https://github.com/RayTale/Ray ）

---

第二小节总结：由上往下，架构层面粗粒度层面表达了采用Actor模型的好处或缘由。

---

Service Fabric

系统开发完成后Actor要组成集群，系统在集群中部署，实现高性能、高可用、可伸缩的要求。部署阶段能够选择Service Fabric或者K8S，目的是下降分布式系统部署、管理的难度，同时知足弹性伸缩。

交易所项目能够采用Service Fabric部署，也能够采用K8S，当时K8S还没这么流行，咱们采用了Service Fabric，Service Fabric 是一款微软开源的分布式系统平台，可方便用户轻松打包、部署和管理可缩放的可靠微服务和容器。开发人员和管理员不需解决复杂的基础结构问题，只需专一于实现苛刻的任务关键型工做负荷，即那些可缩放、可靠且易于管理的工做负荷。支持Windows与Linux部署，Windows上的部署文档齐全，但在Linux上官方资料没有。如今推荐K8S。

---

第三小节总结：

1. 借助Service Fabric或K8S实现低成本运维、构建集群的目的。
2. 创建分布式系统的两种最佳实践：

• 进程级别：容器+运维工具（k8s/sf）
• 线程级别：Actor+运维工具（k8s/sf）

---

上面是我对今天话题的分享。

参考：

1. ES/CQRS部份内容参考：《领域模型 + 内存计算 + 微服务的协奏曲：乾坤（演讲稿）》 2017年互联网应用架构实战峰会
2. 其余细节来自互联网，不一一列出

---

讨论

T: 1000W用户，购买200W SKU，若是不考虑热点SKU，则每一个SKU平均为5个并发减库存的更新； 而总共的SKU分10个数据库存储，则每一个库存储20W SKU。因此20W * 5 = 100W个并发的减库存；

T: 每一个库负责100W的并发更新，这个并发量，不论是否采用actor/es，都要采用group commit的技术

T: 不然单机都不可能达到100W/S的数据写入。

T: 采用es的方式，就是每秒插入100W个事件；不采用ES，就是每秒更新100W次商品减库存的SQL update语句

Y: 哦

T: 不过实际上，除了阿里的体量，不可能并发达到1000W的

T: 1000W用户不表明1000W并发

T: 若是真的是1000W并发，可能实际在线用户至少有10亿了

T: 由于若是只有1000W在线用户，那是不可能这些用户同时在同一秒内发起购买的，你们想一下是否是这样

Y: 这么熟的名字

T: 因此，1000W在线用户的并发实际只有10W最多了

T: 也就是单机只有1W的并发更新，不须要group commit也无压力

Y: 嗯

问答

Q1：单点故障后，正在处理的 cache 数据如何处理的，例如，http,tcp请求…毕竟涉及到钱

A：actor有激活和失活的生命周期，激活的时候使用快照和Events来恢复最新内存状态，失活的时候保存快照。actor框架保证系统中同一个key只会存在同一个actor，当单点故障后，actor会在其它节点重建并恢复最新状态。

Q2：event ID生成的速度如何保证有效的scale？有没有遇到须要后期插入一些event，修正前期系统运行的bug？有没有遇到须要把前期已经定好的event再拆细的状况？有遇到系统错误，须要replay event的状况？ A：1. 当时项目中event ID采用了MongoDB的ObjectId生成算法，没有遇到问题；有遇到后期插入event修正以前bug的状况；有遇到将已定好的event修改的状况，采用的方式是加版本号；没有，遇到过系统从新迁移删除快照从新replay event的状况。

Q3：数据落地得策略是什么？仍是说就是直接落地？ A：event数据直接落地；用于支持查询的数据，是Handler消费event后异步落库。

Q4：actor跨物理机器集群事务怎么处理？ A：结合事件溯源，采用最终一致性。

Q5：Grain Persistence使用Relational Storage容量和速度会不会是瓶颈？ A：Grain Persistence存的是Grain的快照和event，event是只增的，速度没有出现瓶颈，并且开源版本测试中PostgreSQL性能优于MongoDB，在存储中针对这两个方面作了优化：好比分表、归档处理、快照处理、批量处理。

Q6：SF中的reliable collection对应到k8s是什么？ A：很差意思，这个我不清楚。

Q7：开发语言是erlang吗？Golang有这样的开发模型库支持吗？ A：开发语言是C#。Golang我了解的很少，proto.actor能够了解一下：https://github.com/AsynkronIT/protoactor-go

Q8：可否来几篇博客阐述如何一步步使用orleans实现一个简单的事件总线 A：事件总线的实现使用的是RabbitMQ，这个能够看一下开源版本的源码EventBus.RabbitMQ部分，博客的可能后面会写，若是不996的话（笑脸）

Q9：每一个pod的actor都不同，如何用k8s部署actor，失败的节点如何监控，并借助k8s自动恢复？ A：actor是无状态的，失败恢复依靠从新激活时事件溯源机制。k8s部署actor官方有支持，能够参考官方示例。在实际项目中使用k8s部署Orleans，我没有实践过，后来有同事验证过能够，具体如何监控不清楚。

Q10：Orleans中，持久化事件时，是否有支持并发冲突的检测，是如何实现的？ A：Orleans不支持；工做中，在事件持久化时作了这方面的工做，方式是根据版本号。

Q11：Orleans中，如何判断消息是否重复处理的？由于分布式环境下，同一个消息可能会被重复发送到actor mailbox中的，而actor自己没法检测消息是否重复过来。 A：是的，在具体项目中，经过框架封装实现了幂等性控制，具体细节是经过插入事件的惟一索引。

Q12：同一个actor是否会存在于集群中的多台机器？若是可能，怎样的场景下可能会出现这种状况？ A：一个Id对应的Actor只会在集群种存在一个。

Q13： 响应式架构 消息模式Actor实现与Scala.Akka应用集成 这本书对理解actor的帮助大吗，还有实现领域驱动设计这本

A：这本书我看过，刚接触这个项目时看的，文章说的有些深奥，由于当时关注的是Orleans，文中讲的是akka，帮助不大，推荐具体项目的官方文档。实现领域驱动这本书有收获，推荐专题式阅读，DDD多在社区交流。