Reactive programming 中如何沿用 Domain Design 的设计方式？

保留领域设计，让 I/O 接口明确表达异步

领域实体、Repository 接口和外部实现的设计可以继续使用。采用 reactive model 后，涉及异步 I/O 的接口契约需要相应调整：查询结果如果不能立即取得，接口就应明确表达这一点。

对于 get(cityName) 这样的单次查询，可以返回 CompletionStage<Optional<City>>，也可以使用所选响应式库的单结果类型。领域实体仍可保持为普通 Java 对象，业务方法继续同步执行。

具体框架的 Handler、数据库记录和 HTTP 响应对象应留在基础设施适配器里。整体可以按六边形架构中的 Ports and Adapters 来组织。

异步执行不改变领域的依赖方向

原来的设计已经做了依赖倒置：

• 内层定义 CitiesRepository 接口，描述所需的能力。
• 外层实现接口，负责访问数据库、HTTP 服务或缓存。
• 内层代码通过接口获取 City，无需了解数据来源。

异步执行不会改变这条依赖方向。Repository 接口是访问边界，可以表达操作如何完成；它与领域实体的职责不同。

这里需要分清两个问题：

1. 数据模型依赖什么？City 不应依赖数据库客户端或框架的 Handler。
2. 调用方如何获得结果？异步查询需要通过回调、future 或响应式类型传递结果和失败。

返回 CompletionStage<City> 会引入对 Java 异步抽象的依赖，但不会引入对某个数据库或 Vert.x 的依赖。如果返回第三方库的响应式类型，接口就会依赖该库。选择返回类型时，需要接受相应的架构取舍。

在 reactive programming 中，方法也可以有返回值：它可以立即返回一个对象，表示未来的结果或后续的数据流。

第一步：按操作语义选择返回类型

get(cityName) 通常只产生一个结果，需要先确定如何表示“没有找到城市”。例如，可以约定：

• 找到城市：正常完成，结果为 Optional.of(city)。
• 没有找到：正常完成，结果为 Optional.empty()。
• 查询失败：异常完成。

接口可以写成：

package domain;

import domain.beans.City;
import java.util.Optional;
import java.util.concurrent.CompletionStage;

public interface CitiesRepository {
    CompletionStage<Optional<City>> findByName(String cityName);
}

findByName 更容易让人理解这个查询可能找不到结果。也可以保留 get，但要写清它的行为约定。

CompletionStage 适合表示单次异步结果，也支持后续组合。它不等同于完整的响应式流抽象，不提供背压协议。

如果需要持续接收城市变更，或逐条读取大量结果，再考虑流类型。Observable、Publisher 等类型在完成、取消和背压方面的语义并不完全相同，选型时要检查具体库的契约。

第二步：在适配器里集中转换框架回调

适配器负责将框架回调转换为内层接口使用的异步类型，应用服务就不必反复处理框架的 Handler。

示例中的 client.find 是示意接口，回调参数约定为 (row, error)，其中 row == null 表示未找到。mapToCity 负责将外部数据转换为领域对象。它们不对应某个具体框架的 API。

@Override
public CompletionStage<Optional<City>> findByName(String cityName) {
    CompletableFuture<Optional<City>> result =
            new CompletableFuture<>();

    try {
        client.find(cityName, (row, error) -> {
            if (error != null) {
                result.completeExceptionally(error);
                return;
            }

            try {
                Optional<City> city = row == null
                        ? Optional.empty()
                        : Optional.of(mapToCity(row));

                result.complete(city);
            } catch (RuntimeException mappingError) {
                result.completeExceptionally(mappingError);
            }
        });
    } catch (RuntimeException invocationError) {
        result.completeExceptionally(invocationError);
    }

    return result;
}

这段代码处理了查询失败、数据转换失败，以及发起调用时同步抛出的异常。实际项目中，还可以在适配器里把客户端异常转换为内层约定的异常，避免上层依赖数据库或 HTTP 客户端的异常类型。

如果客户端已经返回可组合的异步类型，通常可以直接使用它提供的转换能力，无需手写这段桥接代码。

第三步：应用层组合异步操作，领域对象处理业务规则

应用服务在数据就绪后调用领域行为，并安排后续持久化。领域实体只处理已经取得的数据。

例如：

return citiesRepository.findByName(cityName)
        .thenApply(city -> city.orElseThrow(
                () -> new IllegalArgumentException(
                        "Unknown city: " + cityName
                )
        ));

将“城市不存在”作为失败处理，是调用方的业务选择。Repository 仍然区分“未找到”和“查询失败”。

后续如果只做内存计算，可以使用 thenApply。如果下一步要调用返回 CompletionStage<T> 的异步操作，应使用 thenCompose 串联，避免得到嵌套的 CompletionStage<CompletionStage<T>>。

领域对象不必为此把所有方法都改成异步方法。校验城市是否支持某项业务、计算费用等内存中的规则，仍然可以直接返回结果。

如果希望领域模型完全不接触异步类型，也可以把异步查询端口放在应用层。应用服务先取得 City，再传给领域对象或领域服务。接口放在哪一层，应根据使用方和它表达的业务能力来判断，仅凭名称无法决定。

Callback、Promise 与 Observable 如何选择

可以自定义 DomainHandler<City>，但接口不能只传递成功结果，还需要说明如何通知失败、回调是否只执行一次，以及是否允许取消。

多个异步操作串联时，自定义 callback API 往往还需要支持结果转换、失败传播和操作组合。对于单次查询，直接采用已有的 future/promise 抽象通常更省事。

Observable 适合某些流式场景，也能表示单个结果，但它允许接口表达更多行为。对于只查一个城市的操作，使用单结果语义明确的类型，调用方更容易理解。

如果整个应用已经统一使用某个响应式库，应用层端口也可以直接返回该库的类型。为每种库重新设计一套自有异步抽象，还要额外维护组合与取消语义，只有确实需要这层隔离时才值得做。

实现时还要注意什么

返回异步类型并不保证底层非阻塞。如果实现仍在事件循环线程中调用阻塞式数据库驱动，线程照样会被占用。必须使用阻塞式客户端时，应将操作安排到专门处理阻塞工作的执行环境。

不要为了保留原来的 City get(String) 签名，在适配器里调用 get() 或 join() 等待异步结果。这会把等待隐藏在同步接口后面，在事件循环上尤其容易造成问题。

还需要明确以下行为：

• CompletionStage 的非 Async 回调不保证切换到专用线程，回调中应避免阻塞和耗时计算。
• 取消返回对象上的操作，不一定会取消底层数据库查询或 HTTP 请求，需要客户端和适配器共同支持。
• 接口改成异步后，超时、重试和事务仍需单独处理，尤其要确认事务上下文如何跨异步步骤传播。
• 如果多个回调可能并发访问同一可变实体，需要限制执行顺序或明确并发控制方式。

备注：内容仅供参考。