在本系列的第一部分和第二部分中，我介绍了 命令处理器 和 工作单元和仓库模式。我本打算写关于消息总线以及更多关于领域建模的内容，但我需要先快速浏览一下这个。
如果您刚刚开始阅读消息总线部分，并且您来这里是为了了解应用程序控制的标识符，您会在文章末尾找到这些内容，在关于 ORM、CQRS 和一些对初级程序员的随意调侃之后。
什么是 CQS？
命令查询分离 原则最早由 Bertrand Meyer 在 80 年代后期提出。根据 维基百科，该原则指出
每个方法应该要么是一个执行操作的命令，要么是一个向调用者返回数据的查询，但不能两者兼有。换句话说，“提问不应该改变答案”。更正式地说，方法只有在引用透明且因此没有副作用的情况下才应该返回值。
引用透明是函数式编程中的一个重要概念。简而言之，如果可以将函数替换为静态值，则该函数是引用透明的。
class LightSwitch:
def toggle_light(self):
self.light_is_on = not self.light_is_on
return self.light_is_on
@property
def is_on(self):
return self.light_is_on
在这个类中，`is_on` 方法是引用透明的 - 我可以将其替换为 `True` 或 `False` 值，而不会损失任何功能，但是 `toggle_light` 方法是副作用的：用静态值替换其调用将破坏系统的契约。为了遵守命令查询分离原则，我们不应该从 `toggle_light` 方法返回值。
在某些语言中，我们会说 `is_on` 方法是“纯粹的”。将我们的函数分为有副作用的和纯粹的函数的优点是代码变得更容易推理。Haskell 喜欢纯函数，并使用这种可推理性来做奇怪的事情，例如在编译时为您重新排序代码以使其更高效。对于我们这些使用更平实的语言的人来说，如果命令和查询被明确区分，那么我可以通读代码库并了解状态可能发生的所有变化方式。这对于调试来说是一个巨大的胜利，因为没有什么比在您无法弄清楚哪些代码路径正在更改您的数据时排除系统故障更糟糕的了。
我们如何从命令处理程序架构中获取数据？当我们在命令处理程序系统中工作时，我们显然使用命令和处理程序来执行状态更改，但是当我们想要从模型中获取数据时应该怎么办？查询的等效端口是什么？
答案是“视情况而定”。成本最低的选择是在您的 UI 入口点中重用您的仓库。
@app.route("/issues")
def list_issues():
with unit_of_work_manager.start() as unit_of_work:
open_issues = unit_of_work.issues.find_by_status('open')
return json.dumps(open_issues)
这完全没问题，除非您有复杂的格式，或者您的系统有多个入口点。以这种方式直接使用您的仓库的问题在于，这是一个温水煮青蛙的过程。迟早您会遇到一个紧张的期限和一个简单的需求，并且诱惑是跳过所有命令/处理程序的废话，并在 Web API 中直接完成它。
@app.route('/issues/<issue_id>', methods=['DELETE'])
def delete_issue(issue_id):
with unit_of_work_manager.start() as uow:
issue = uow.issues[issue_id]
issue.delete()
uow.commit()
超级方便，但是您需要添加一些错误处理和一些日志记录以及电子邮件通知。
@app.route('/issues/<issue_id>', methods=['DELETE'])
def delete_issue(issue_id):
logging.info("Handling DELETE of issue "+str(issue_id))
with unit_of_work_manager.start() as uow:
issue = uow.issues[issue_id]
if issue is None:
logging.warn("Issue not found")
flask.abort(404)
if issue.status != 'deleted':
issue.delete()
uow.commit()
try:
smtp.send_notification(Issue.Deleted, issue_id)
except:
logging.error(
"Failed to send email notification for deleted issue "
+ str(issue_id), exn_info=True)
else:
logging.info("Issue already deleted. NOOP")
return "Deleted!", 202
然后，我们又回到了起点：业务逻辑与粘合代码混合在一起，整个混乱慢慢凝结在我们的 Web 控制器中。当然，温水煮青蛙的论点并不是不做某事的充分理由，因此如果您的查询非常简单，并且您可以避免从控制器进行更新的诱惑，那么您不妨继续从仓库读取，一切都很好，我为您祝福。如果您想避免这种情况，因为您的读取很复杂，或者因为您试图保持纯粹，那么我们可以显式定义我们的视图。
class OpenIssuesList:
def __init__(self, sessionmaker):
self.sessionmaker = sessionmaker
def fetch(self):
with self.sessionmaker() as session:
result = session.execute(
'SELECT reporter_name, timestamp, title
FROM issues WHERE state="open"')
return [dict(r) for r in result.fetchall()]
@api.route('/issues/')
def list_issues():
view_builder = OpenIssuesList(session_maker)
return jsonify(view_builder.fetch())
这是我最喜欢向初级程序员教授端口和适配器的地方，因为对话不可避免地会像这样进行
面容光滑的年轻人：哇，嗯……您是 - 我们只是要将 SQL 硬编码到那里吗？只是……在数据库上运行它？
饱经风霜的老架构师：是的，我想是这样。做最简单可行的事情，对吧？ YOLO，等等。
sfy：哦，好的。嗯……但是工作单元和领域模型以及服务层和六边形的东西呢？您不是说“数据访问应该针对用例的聚合根执行，以便我们保持对事务边界的严格控制”吗？
goa：呃……我现在不想做那个，我想我饿了。
sfy：对，对……但是如果您的数据库模式更改了怎么办？
goa：我想我会回来并更改那一行的 SQL。如果我忘记了，我的验收测试会失败，所以我无法通过 CI 获取代码。
sfy：但是我们为什么不使用我们编写的 Issue 模型呢？仅仅忽略它并返回这个 dict 似乎很奇怪……而且您说过“避免直接依赖框架。针对抽象工作，这样如果您的依赖项发生更改，就不会强制更改波及到您的域”。您知道我们无法对此进行单元测试，对吗？
goa：哈！你是什么，某种架构宇航员吗？领域模型！谁需要它们。
为什么要有一个单独的读取模型？
根据我的经验，团队在使用 ORM 时通常会犯两种错误。最常见的错误是没有足够重视其用例的边界。这导致应用程序向数据库发出过多调用，因为人们编写了这样的代码
# Find all users who are assigned this task
# [[and]] notify them and their line manager
# then move the task to their in-queue
notification = task.as_notification()
for assignee in task.assignees:
assignee.manager.notifications.add(notification)
assignee.notifications.add(notification)
assignee.queues.inbox.add(task)
ORM 使通过对象模型“点”的方式非常容易，并假装我们的数据在内存中，但是当 ORM 生成数百个 select 语句作为响应时，这很快会导致性能问题。然后他们对性能感到愤怒，并撰写长篇博客文章，内容是 ORM 有多糟糕，是一种反模式，只有 n00b 才喜欢它。这类似于将 OO 归咎于您的域逻辑最终出现在控制器中。
团队犯的第二个错误是在他们不需要 ORM 时使用 ORM。我们首先为什么要使用 ORM？我认为好的 ORM 为我们提供了两件事
- 工作单元模式，可用于控制我们的一致性边界。
- 数据映射器模式，使我们可以将复杂的对象图映射到关系表，而无需编写大量的无聊的粘合代码。
总而言之，这些模式通过消除所有数据库的繁文缛节来帮助我们编写丰富的领域模型，因此我们可以专注于我们的用例。这使我们能够以内部一致的方式对复杂的业务流程进行建模。但是，当我编写 GET 方法时，我并不关心这些。我的视图不需要任何业务逻辑，因为它不会更改任何状态。对于 99.5% 的用例，即使我的数据是在事务内部获取的也无关紧要。如果我在列出问题时执行脏读，则可能会发生以下三种情况之一
- 我可能会看到尚未提交的更改 - 也许刚刚删除的问题仍会显示在列表中。
- 我可能看不到已提交的更改 - 问题可能会从列表中丢失，或者标题可能会过时 10 毫秒。
- 我可能会看到数据的重复项 - 问题可能会在列表中出现两次。
在许多系统中，所有这些情况都不太可能发生，并且会通过页面刷新或单击链接以查看更多数据来解决。需要明确的是，我不是建议您关闭 SELECT 语句的事务，只是指出事务一致性通常只是在我们更改状态时才是一个真正的要求。在查看状态时，我们几乎总是可以接受较弱的一致性模型。
CQRS 是系统级的 CQS
CQRS 代表命令查询职责分离，它是由 Greg Young 推广的一种架构模式。很多人误解了 CQRS，并认为您需要使用单独的数据库和疯狂的异步处理器才能使其工作。您可以做这些事情，我稍后想写更多关于这方面的内容，但是 CQRS 只是意味着我们分离了写入模型（我们通常认为的领域模型）和读取模型（一个轻量级的简单模型，用于在 UI 上显示或回答有关域状态的问题）。
当我处理写入请求（命令）时，我的工作是保护系统的不变性，并按照领域专家脑海中出现的方式对业务流程进行建模。我采纳我们的业务分析师的集体理解，并将其转化为一个状态机，使其能够完成有用的工作。当我处理读取请求（查询）时，我的工作是以尽可能快的速度从数据库中获取数据并将其显示在屏幕上，以便用户可以查看它。任何妨碍我这样做的事情都是臃肿的。
这不是一个新想法，也不是特别有争议。我们都尝试过针对 ORM 编写报告或复杂的层次结构列表页面，并遇到了性能瓶颈。当我们达到那个程度时，我们唯一能做的 - 除了重写整个模型或放弃使用 ORM 之外 - 就是用原始 SQL 重写我们的查询。曾经有一段时间，我会因为这样做而感到难过，就好像我在作弊一样，但是现在我只是认识到，我的查询的要求与我的命令的要求从根本上是不同的。
对于系统的写入端，使用 ORM；对于读取端，使用任何 a) 快速且 b) 方便的东西。
应用程序控制的标识符
此时，一位非初级程序员会说
好吧，聪明的架构师先生，如果我们的命令不能返回任何值，并且我们的领域模型对数据库一无所知，那么我如何从我的保存方法中获取 ID？假设我创建一个用于创建新问题的 API，并且当我 POST 了新问题时，我想将用户重定向到一个他们可以 GET 其新问题的端点。我如何才能取回 ID？
我建议您处理此问题的方式很简单 - 不要让您的数据库为您选择 ID，而只需自己选择它们。
@api.route('/issues', methods=['POST'])
def report_issue(self):
# uuids make great domain-controlled identifiers, because
# they can be shared amongst several systems and are easy
# to generate.
issue_id = uuid.uuid4()
cmd = ReportIssueCommand(issue_id, **request.get_json())
handler.handle(cmd)
return "", 201, { 'Location': '/issues/' + str(issue_id) }
有很多方法可以做到这一点，最常见的方法是只使用 UUID，但是您也可以实现类似 hi-lo 的东西。在新的 代码示例 中，我实现了三个 Flask 端点，一个用于创建新问题，一个用于列出所有问题，一个用于查看单个问题。我正在使用 UUID 作为我的标识符，但我仍然在 issues 表上使用整数主键，因为在聚集索引中使用 GUID 会导致表碎片和 悲伤 。
好的，快速检查 - 我们与最初的端口和适配器图相比如何？这些概念如何映射？
非常好！我们的领域是纯粹的，并且对基础设施或 IO 一无所知。我们有一个命令和一个处理程序来编排用例，我们可以从测试或 Flask 驱动我们的应用程序。最重要的是，外层的层依赖于中心层的层。
下次我将回到讨论消息总线。