六边形架构模式

意图

六边形架构模式（也称为端口和适配器模式）由 Alistair Cockburn 博士在 2005 年提出。它旨在创建松耦合架构，在这种架构中，可以独立测试应用程序组件，而不依赖于数据存储或用户界面（UI）。此模式有助于防止数据存储和用户界面的技术锁定。这样一来，随着时间的推移，技术堆栈的更改就变得更加容易，对业务逻辑的影响也有限或没有影响。在此松耦合架构中，应用程序通过称为端口的接口与外部组件通信，并使用适配器来转换与这些组件的技术交互。

动机

六边形架构模式用于将业务逻辑（域逻辑）与相关的基础设施代码（例如用于访问数据库或外部 API 的代码）隔离。此模式对于为需要与外部服务集成的 AWS Lambda 函数创建松耦合业务逻辑和基础设施代码非常有用。在传统架构中，常见的做法是将业务逻辑作为存储过程嵌入数据库层，并嵌入到用户界面中。此做法再加上在业务逻辑内使用特定于用户界面的结构，会导致紧密耦合的架构，从而在数据库迁移和用户体验（UX）现代化工作中造成瓶颈。六边形架构模式使您能够按目的（而不是按技术），来设计系统和应用程序。此策略可以生成易于交换的应用程序组件，例如数据库、UX 和服务组件。

适用性

在以下情况下使用六边形架构模式：

您想解耦应用程序架构，以创建可以全面测试的组件。
多种类型的客户端可以使用相同的域逻辑。
您的用户界面和数据库组件需要定期进行技术更新，而不会影响应用程序逻辑。
您的应用程序需要多个输入提供程序和输出使用者，而自定义应用程序逻辑会导致代码复杂和缺乏可扩展性。

问题和注意事项

领域驱动型设计：六边形架构特别适用于领域驱动设计（DDD）。每个应用程序组件都代表 DDD 中的一个子域，六边形架构可用于实现应用程序组件之间的松耦合。
可测试性：根据设计，六边形架构使用抽象作为输入和输出。因此，由于固有的松耦合，编写单元测试和单独测试变得更加容易。
复杂性：将业务逻辑与基础设施代码隔离的复杂性，如果处理得当，可以带来巨大的好处，例如敏捷性、测试覆盖率和技术适应性。否则，问题可能会变得复杂而难以解决。
维护开销：只有当应用程序组件需要多个输入源和输出目标进行写入时，或者输入和输出数据存储必须随着时间的推移而发生变化时，才有必要添加使架构可插拔的适配器代码。否则，适配器将成为另一个需要维护的额外层，这会带来维护开销。
延迟问题：使用端口和适配器会增加一层，这可能会导致延迟。

实现

六边形架构支持将应用程序和业务逻辑与基础设施代码以及将应用程序与 UI、外部 API、数据库和消息代理集成的代码隔离。您可以通过端口和适配器，轻松地将业务逻辑组件连接到应用程序架构中的其他组件（例如数据库）。

端口是应用程序组件中与技术无关的入口点。这些自定义接口决定了允许外部参与者与应用程序组件通信的接口，而不管是谁或什么实现了该接口。这与 USB 端口允许许多不同类型的设备与计算机通信的原理类似，只要它们使用 USB 适配器即可。

适配器使用特定技术通过端口与应用程序进行交互。适配器插入这些端口，从端口接收数据或向端口提供数据，然后转换数据以进行进一步处理。例如，REST 适配器使参与者能够通过 REST API 与应用程序组件进行通信。一个端口可以具有多个适配器，而不会对端口或应用程序组件造成任何风险。为了扩展上一个示例，向同一端口添加 GraphQL 适配器为参与者提供了一种额外的方式，可通过 GraphQL API 与应用程序交互，而不会影响 REST API、端口或应用程序。

端口连接到应用程序，适配器用作与外界的连接。您可以使用端口创建松耦合的应用程序组件，并通过更改适配器来交换依赖组件。这使应用程序组件无需任何上下文感知即可与外部输入和输出进行交互。组件可在任何级别进行互换，有助于自动化测试。您可以独立测试组件，而无需依赖基础设施代码，也无需预调配整个环境来进行测试。应用程序逻辑不依赖于外部因素，因此测试得以简化，模拟依赖关系也变得更加容易。

例如，在松耦合架构中，应用程序组件应该能够在不知道数据存储详细信息的情况下读取和写入数据。应用程序组件的职责是向接口（端口）提供数据。适配器定义写入数据存储的逻辑，数据存储可以是数据库、文件系统或对象存储系统（例如 Amazon S3），具体取决于应用程序的需求。

高级架构

应用程序或应用程序组件包含核心业务逻辑。它从端口接收命令或查询，并通过端口向外部参与者发送请求，这些请求是通过适配器实现的，如下图所示。

使用 AWS 服务进行实施

AWS Lambda 函数通常包含业务逻辑和数据库集成代码，这两者紧密耦合以实现目标。您可以使用六边形架构模式将业务逻辑与基础设施代码隔离。此隔离使得业务逻辑可以进行单元测试，而无需依赖数据库代码，从而提高了开发过程的敏捷性。

在以下架构中，Lambda 函数实现六边形架构模式。Lambda 函数由 Amazon API Gateway REST API 发起。该函数实现业务逻辑并将数据写入 DynamoDB 表。

代码示例

本节中的示例代码展示了如何使用 Lambda 实现域模型，将其与基础设施代码（例如用于访问 DynamoDB 的代码）隔离，并为该函数实现单元测试。

域模型

域模型类别对外部组件或依赖关系一无所知，它只实现业务逻辑。在以下示例中，类别 Recipient 是一个域模型类别，用于检查预留日期是否存在重叠。


class Recipient:
    def __init__(self, recipient_id:str, email:str, first_name:str, last_name:str, age:int):
        self.__recipient_id = recipient_id
        self.__email = email
        self.__first_name = first_name
        self.__last_name = last_name
        self.__age = age
        self.__slots = []
 
    @property
    def recipient_id(self):
        return self.__recipient_id
    #.....     
 
    def are_slots_same_date(self, slot:Slot) -> bool:
        for selfslot in self.__slots:
            if selfslot.reservation_date == slot.reservation_date:
                return True        
        return False
 
    def is_slot_counts_equal_or_over_two(self) -> bool:
    #.....

输入端口

RecipientInputPort 类别连接到接收者类别并运行域逻辑。


class RecipientInputPort(IRecipientInputPort):
    def __init__(self, recipient_output_port: IRecipientOutputPort, slot_output_port: ISlotOutputPort):
        self.__recipient_output_port = recipient_output_port
        self.__slot_output_port = slot_output_port

    '''
    make reservation: adapting domain model business logic
    '''
    def make_reservation(self, recipient_id:str, slot_id:str) -> Status:
        status = None        
        
        # ---------------------------------------------------
        # get an instance from output port
        # ---------------------------------------------------
        recipient = self.__recipient_output_port.get_recipient_by_id(recipient_id)
        slot = self.__slot_output_port.get_slot_by_id(slot_id)

        if recipient == None or slot == None:
            return Status(400, "Request instance is not found. Something wrong!")

        print(f"recipient: {recipient.first_name}, slot date: {slot.reservation_date}")

        # ---------------------------------------------------
        # execute domain logic
        # ---------------------------------------------------
        ret = recipient.add_reserve_slot(slot)

        # ---------------------------------------------------
        # persistent an instance throgh output port
        # ---------------------------------------------------
        if ret == True:
            ret = self.__recipient_output_port.add_reservation(recipient)

        if ret == True:
            status = Status(200, "The recipient's reservation is added.")
        else:
            status = Status(200, "The recipient's reservation is NOT added!")
        return status

DynamoDB 适配器类别

DDBRecipientAdapter 类别实现对 DynamoDB 表的访问。


class DDBRecipientAdapter(IRecipientAdapter):
    def __init__(self):
        ddb = boto3.resource('dynamodb')
        self.__table = ddb.Table(table_name)
 
    def load(self, recipient_id:str) -> Recipient:
        try:
            response = self.__table.get_item(
                Key={'pk': pk_prefix + recipient_id})
      　... 
 
    def save(self, recipient:Recipient) -> bool:
        try:
            item = {
                "pk": pk_prefix + recipient.recipient_id,
                "email": recipient.email,
                "first_name": recipient.first_name,
                "last_name": recipient.last_name,
                "age": recipient.age,
                "slots": []
            }
          # ...

Lambda 函数 get_recipient_input_port 是 RecipientInputPort 类别实例的工厂。它使用相关的适配器实例构造输出端口类别的实例。


def get_recipient_input_port():
    return RecipientInputPort(
        RecipientOutputPort(DDBRecipientAdapter()), 
        SlotOutputPort(DDBSlotAdapter()))
 
def lambda_handler(event, context):

    body = json.loads(event['body'])
    recipient_id = body['recipient_id']
    slot_id = body['slot_id']
 
    # get an input port instance
    recipient_input_port = get_recipient_input_port()
    status = recipient_input_port.make_reservation(recipient_id, slot_id)
 
    return {
        "statusCode": status.status_code,
        "body": json.dumps({
            "message": status.message
        }),
    }

单元测试

您可以通过注入模拟类别来测试域模型类别的业务逻辑。以下示例提供了域模型 Recipent 类别的单元测试。


def test_add_slot_one(fixture_recipient, fixture_slot):
    slot = fixture_slot
    target = fixture_recipient
    target.add_reserve_slot(slot)
    assert slot != None
    assert target != None
    assert 1 == len(target.slots)
    assert slot.slot_id == target.slots[0].slot_id
    assert slot.reservation_date == target.slots[0].reservation_date
    assert slot.location == target.slots[0].location
    assert False == target.slots[0].is_vacant
 
def test_add_slot_two(fixture_recipient, fixture_slot, fixture_slot_2):
    #.....
 
def test_cannot_append_slot_more_than_two(fixture_recipient, fixture_slot, fixture_slot_2, fixture_slot_3):
    #.....
 
def test_cannot_append_same_date_slot(fixture_recipient, fixture_slot):
    #.....

GitHub 存储库

有关此模式示例架构的完整实施，请参阅 GitHub 存储库 https://github.com/aws-samples/aws-lambda-domain-model-sample。

视频

以下视频（日语）讨论了如何使用 Lambda 函数在实现域模型时使用六边形架构。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

文档惯例

事件溯源模式

发布-订阅模式