DDD 这个术语来自 Eric Evans 的著作：《领域驱动设计：应对软件核心的复杂性》。在他的书中，他描述了一系列旨在帮助我们构建可维护、丰富的软件系统以解决客户问题的实践。这本书有 560 页内容丰富的见解，如果我的总结遗漏了一些细节，请您原谅，但简而言之，他建议
- 非常仔细地倾听您的领域专家——您正在用软件自动化或协助其工作的人员。
- 学习他们使用的行话，并帮助他们提出新的行话，以便他们心智模型中的每个概念都用一个精确的术语来命名。
- 使用这些术语来建模您的软件；领域专家的名词和动词应该是您在建模中使用的类和方法。
- 每当您对领域的共同理解出现差异时，请再次与领域专家交谈，然后积极地进行重构。
这在理论上听起来很棒，但在实践中，我们经常发现我们的业务逻辑从我们的模型对象中逃逸；我们最终让逻辑渗入控制器或臃肿的“管理器”类。我们发现重构变得困难：我们无法拆分一个大型且重要的类，因为这会严重影响数据库模式；或者我们无法重写算法的内部结构，因为它已经与为不同用例而存在的代码紧密耦合。好消息是，这些问题是可以避免的，因为它们是由代码库中缺乏组织性引起的。事实上，DDD 书籍的一半内容都在讲述解决这些问题的工具，但可能很难理解如何在完整系统的上下文中一起使用它们。
我想用本系列文章介绍一种名为 端口和适配器 的架构风格，以及一种名为 命令处理器 的设计模式。我将用 Python 解释这些模式，因为这是我日常使用的语言，但这些概念适用于任何 OO 语言，并且可以进行调整以在函数式上下文中完美运行。可能比您习惯的层次和抽象要多得多，特别是如果您来自 Django 背景或类似情况，但请耐心等待。为了在一开始构建更复杂的系统，我们可以避免以后的大部分 偶然复杂性。
我们将要构建的系统是一个问题管理系统，供服务台使用。我们将要替换一个现有的系统，该系统由一个发送电子邮件的 HTML 表单组成。电子邮件进入一个邮箱，服务台人员会浏览邮件，对问题进行分类，并挑选他们可以解决的问题。有时问题会被长期忽视，服务台团队发明了一套复杂的便利贴和白板布局系统来跟踪正在进行的工作。在一段时间内，这个系统运行良好，但随着系统变得越来越繁忙，裂缝开始显现。
我们与领域专家的第一次对话：“流程的第一步是什么？”你问，“工单是如何进入邮箱的？”。
“嗯，首先发生的事情是用户访问网页，他们填写一些详细信息并报告问题。这会将一封电子邮件发送到问题日志中，然后我们每天早上从日志中挑选问题”。
“那么当用户报告问题时，您需要从他们那里获取的最小数据集是什么？”
“我们需要知道他们是谁，所以他们的姓名，以及我想是电子邮件。嗯……还有问题描述。他们应该添加一个类别，但他们从不这样做，而且我们曾经有一个优先级，但每个人都将他们的问题设置为“极其紧急”，所以它毫无用处。”
“但是类别和优先级可以帮助您对事物进行分类吗？”
“是的，如果我们能让用户正确设置它们，那将非常有帮助。”
这为我们提供了第一个用例：作为用户，我希望能够报告新问题。
好的，在我们开始编写代码之前，让我们谈谈架构。软件系统的架构是整体结构——组织代码并满足我们约束的语言、技术和设计模式的选择 [https://en.wikipedia.org/wiki/Non-functional_requirement]。对于我们的架构，我们将尝试坚持三个原则
- 我们将始终定义用例的开始和结束位置。我们不会有散布在整个代码库中的业务流程。
- 我们将依赖抽象 [https://en.wikipedia.org/wiki/Dependency_inversion_principle]，而不是具体的实现。
- 我们将把胶水代码与业务逻辑区分开来，并将它们放在适当的位置。
首先，我们从领域模型开始。领域模型封装了我们对问题的共同理解，并使用了我们与领域专家约定的术语。为了遵守原则 #2，我们将为任何基础设施或技术问题定义抽象，并在我们的模型中使用这些抽象。例如，如果我们需要发送电子邮件或将实体保存到数据库，我们将通过捕获我们意图的抽象来执行此操作。在本系列中，我们将为我们的领域模型创建一个单独的 python 包，以便我们可以确保它不依赖于系统的其他层。严格遵守此规则将使测试和重构我们的系统变得更容易，因为我们的领域模型不会与数据库和 http 调用的混乱细节纠缠在一起。
在我们的领域模型之外，我们放置了服务。这些是无状态对象，用于对领域执行操作。特别是，对于此系统，我们的命令处理程序是服务层的一部分。
最后，我们有适配器层。此层包含驱动服务层或为领域模型提供服务的代码。例如，我们的领域模型可能具有用于与数据库对话的抽象，但适配器层提供了具体的实现。其他适配器可能包括 Flask API、我们的单元测试集或 celery 事件队列。所有这些适配器都将我们的应用程序连接到外部世界。
为了遵守我们的第一个原则，我们将为此用例定义一个边界，并创建我们的第一个命令处理器。命令处理器是编排业务流程的对象。它执行获取正确对象并在其上调用正确方法的枯燥工作。它类似于 MVC 架构中控制器的概念。
首先，我们创建一个命令对象。
class ReportIssueCommand(NamedTuple):
reporter_name: str
reporter_email: str
problem_description: str
命令对象是一个小对象，它表示系统中可能发生的改变状态的操作。命令没有行为，它们是纯粹的数据结构。您没有理由必须用类来表示它们，因为它们所需要的只是一个名称和一个数据包，但 NamedTuple 是简单性和便利性之间的一个不错的折衷方案。命令是来自外部代理（用户、cron 作业、另一个服务等）的指令，并且以祈使语气命名，例如
- 报告问题
- 准备上传 URI
- 取消未完成的订单
- 从购物车中移除商品
- 打开登录会话
- 下达客户订单
- 开始支付流程
我们应该尽量避免动词“创建”、“更新”或“删除”（及其同义词），因为这些都是技术实现。当我们倾听我们的领域专家时，我们经常发现对于我们试图建模的操作有一个更好的词。如果您的所有命令都命名为“CreateIssue”、“UpdateCart”、“DeleteOrders”，那么您可能没有足够关注您的利益相关者正在使用的语言。
命令对象属于领域，它们表达了您领域的 API。如果每个改变状态的操作都通过命令处理器执行，那么命令列表就是您的领域模型中支持的完整操作列表。这有两个主要好处
- 如果系统中更改状态的唯一方法是通过命令，那么命令列表会告诉我我需要测试的所有内容。没有其他代码路径可以修改数据。
- 因为我们的命令是轻量级的、无逻辑的对象，所以我们可以从 HTTP post、celery 任务、命令行 csv 读取器或单元测试中创建它们。它们为我们的系统形成了一个简单而稳定的 API，该 API 不依赖于任何实现细节，并且可以通过多种方式调用。
为了处理我们的新命令，我们需要创建一个命令处理器。
class ReportIssueCommandHandler:
def __init__(self, issue_log):
self.issue_log = issue_log
def __call__(self, cmd):
reported_by = IssueReporter(
cmd.reporter_name,
cmd.reporter_email)
issue = Issue(reported_by, cmd.problem_description)
self.issue_log.add(issue)
命令处理器是编排系统行为的无状态对象。它们是一种胶水代码，管理着获取和保存对象，然后通知系统其他部分的枯燥工作。为了遵守原则 #3，我们将此代码保存在一个单独的层中。为了满足原则 #1，每个用例都是一个单独的命令处理器，并且具有明确定义的开始和结束。每个命令都由恰好一个命令处理器处理。
一般来说，所有命令处理器都具有相同的结构
- 从我们的持久存储中获取当前状态。
- 更新当前状态。
- 持久化新状态。
- 通知任何外部系统我们的状态已更改。
我们通常会避免在我们的处理程序中使用 if 语句、循环和其他此类魔法，并坚持一条可能的执行线。命令处理器是枯燥的胶水代码。由于我们的命令处理器只是胶水代码，我们不会在其中放入任何业务逻辑——它们不应该做出任何业务决策。例如，让我们稍微跳到一个新的命令处理器
class MarkIssueAsResolvedHandler:
def __init__(self, issue_log):
self.issue_log = issue_log
def __call__(self, cmd):
issue = self.issue_log.get(cmd.issue_id)
# the following line encodes a business rule
if (issue.state != IssueStatus.Resolved):
issue.mark_as_resolved(cmd.resolution)
这个处理程序违反了我们的胶水代码原则，因为它编码了一个业务规则：“如果问题已经解决，那么它不能再次被解决”。这条规则属于我们的领域模型，可能在我们的 Issue 对象的 mark_as_resolved 方法中。我倾向于为我的命令处理程序使用类，并使用 call 魔术方法调用它们，但函数作为处理程序也是完全有效的。首选类的主要原因是它可以使依赖项管理更容易一些，但这两种方法是完全等效的。例如，我们可以像这样重写我们的 ReportIssueHandler
def ReportIssue(issue_log, cmd):
reported_by = IssueReporter(
cmd.reporter_name,
cmd.reporter_email)
issue = Issue(reported_by, cmd.problem_description)
issue_log.add(issue)
如果魔术方法让您感到不适，您可以将处理程序定义为一个公开 handle 方法的类，如下所示
class ReportIssueHandler:
def handle(self, cmd):
...
无论您如何构建它们，命令和处理程序的重要思想都是
- 命令是无逻辑的数据结构，具有名称和一堆值。
- 它们形成了一个稳定、简单的 API，描述了我们的系统可以做什么，并且不依赖于任何实现细节。
- 每个命令只能由一个处理程序处理。
- 每个命令都指示系统运行一个用例。
- 处理程序通常会执行以下步骤：获取状态、更改状态、持久化状态、通知其他方状态已更改。
让我们看一下完整的系统，我将所有文件连接到一个代码清单中，以便于理解，但在 git 存储库 [https://github.com/bobthemighty/blog-code-samples/tree/master/ports-and-adapters/01] 中，我将系统的各个层拆分为单独的包。在现实世界中，我可能会为整个应用程序使用一个 python 包，但在其他语言（Java、C#、C++）中，我通常会为每一层使用一个二进制文件。以这种方式拆分包使理解依赖项如何工作变得更容易。
from typing import NamedTuple
from expects import expect, have_len, equal
# Domain model
class IssueReporter:
def __init__(self, name, email):
self.name = name
self.email = email
class Issue:
def __init__(self, reporter, description):
self.description = description
self.reporter = reporter
class IssueLog:
def add(self, issue):
pass
class ReportIssueCommand(NamedTuple):
reporter_name: str
reporter_email: str
problem_description: str
# Service Layer
class ReportIssueHandler:
def __init__(self, issue_log):
self.issue_log = issue_log
def __call__(self, cmd):
reported_by = IssueReporter(
cmd.reporter_name,
cmd.reporter_email)
issue = Issue(reported_by, cmd.problem_description)
self.issue_log.add(issue)
# Adapters
class FakeIssueLog(IssueLog):
def __init__(self):
self.issues = []
def add(self, issue):
self.issues.append(issue)
def get(self, id):
return self.issues[id]
def __len__(self):
return len(self.issues)
def __getitem__(self, idx):
return self.issues[idx]
email = "bob@example.org"
name = "bob"
desc = "My mouse won't move"
class When_reporting_an_issue:
def given_an_empty_issue_log(self):
self.issues = FakeIssueLog()
def because_we_report_a_new_issue(self):
handler = ReportIssueHandler(self.issues)
cmd = ReportIssueCommand(name, email, desc)
handler(cmd)
def the_handler_should_have_created_a_new_issue(self):
expect(self.issues).to(have_len(1))
def it_should_have_recorded_the_issuer(self):
expect(self.issues[0].reporter.name).to(equal(name))
expect(self.issues[0].reporter.email).to(equal(email))
def it_should_have_recorded_the_description(self):
expect(self.issues[0].description).to(equal(desc))
这里的功能不多，而且我们的问题日志有一些问题，首先，目前还没有办法查看日志中的问题，其次，每次我们重启进程时，我们都会丢失所有数据。我们将在下一部分 [https://io.made.com/blog/repository-and-unit-of-work-pattern-in-python/] 中解决第二个问题。