《AI时代的程序员修养》前面几篇，从程序、数据结构、执行模型讲到并发。接下来要看一个更贴近日常开发的问题：模块之间怎么说话。

这个问题看起来很普通。一个模块要调用另一个模块，不就是写个函数、发个 HTTP 请求、丢条消息，或者查一下数据库吗？但真实系统里，模块通信方式会决定耦合度、事务边界、失败传播、延迟、吞吐、可观测性和后续演进空间。

AI 很擅长补出调用代码，却经常默认选择最顺手的通信方式。你不明确告诉它边界，它就可能把本该异步的事写成同步调用，把本该有契约的接口写成随手传 dict，把本该隔离的模块通过共享数据库绑死。

通信方式就是系统边界

模块通信不是“怎么把数据传过去”这么简单。每一种通信方式都在回答几组问题：

• 调用方需不需要等结果？
• 被调用方失败时，调用方怎么办？
• 两边是不是必须同时在线？
• 数据是否必须在同一个事务里生效？
• 接口变更时，谁会被影响？
• 调用链能不能被观测和追踪？
• 流量突然变大时，压力落在哪一侧？

常见通信方式可以先放进一张表里：

通信方式	典型场景	优点	主要风险
函数调用	同进程模块、领域服务、工具函数	简单、快速、类型和测试容易控制	模块边界容易被写穿，隐式依赖变多
HTTP / REST	跨服务业务接口、开放 API	通用、易调试、生态成熟	延迟、超时、版本兼容、错误码设计
RPC / gRPC	内部高频服务调用、强 schema 场景	契约清晰、性能好、代码生成方便	调试成本高，服务治理要求更高
消息队列	异步任务、削峰、解耦、最终一致	调用方不必等，能缓冲流量	幂等、重复消费、顺序、死信复杂
事件流	审计、数据同步、多个消费者读取	可追溯、易扩展消费者	schema 演进、重放、offset 管理复杂
共享数据库	小系统、历史系统、报表读取	实现快，少一层接口	强耦合，绕过业务规则，难拆分
文件 / 对象存储	大文件交换、批量导入导出	适合大对象，成本低	生命周期、权限、清理、原子性要设计

这张表不是让人背，而是提醒自己：每次让 AI 写“调用另一个模块”之前，都应该先选通信方式。选错了，代码越多越麻烦。

函数调用：最简单，也最容易把边界写没

同一个进程里，函数调用是最自然的通信方式。它快、直观、容易测试，也最容易被 AI 写出来。

比如一个订单服务里，AI 可能会生成这样的代码：

def create_order(user_id: str, items: list[OrderItem]) -> Order:
    user = user_service.get_user(user_id)
    price = pricing_service.calculate(items, user.level)
    inventory_service.reserve(items)
    order = order_repository.create(user_id=user_id, items=items, price=price)
    notification_service.send_order_created(user.email, order.id)
    return order

这段代码乍看很清楚，但里面至少混了三类操作：

• 必须在主流程里完成的校验和计算。
• 需要和订单创建保持一致的库存预占。
• 可以异步执行的通知。

函数调用的问题不在函数本身，而在边界太容易消失。所有模块都在同一个进程里，调用成本低，AI 就会很自然地把一切串起来。最后一个业务动作变成一条长链，任何一步慢或失败，都拖住整个请求。

更稳的拆法，是先区分同步核心路径和异步副作用：

def create_order(command: CreateOrderCommand) -> Order:
    user = user_policy.require_active_user(command.user_id)
    price = pricing.calculate(command.items, user.level)

    with transaction():
        inventory.reserve(command.items)
        order = order_repository.create(command.user_id, command.items, price)
        outbox.add("order.created", {"order_id": order.id})

    return order

通知不在这里直接发，而是通过 outbox 或消息队列异步处理。函数调用仍然存在，但它只服务主事务内的业务规则。

让 AI 写模块内调用时，可以明确要求：

请把同步核心路径和异步副作用分开。
主流程只包含：校验、价格计算、库存预占、订单创建。
通知、统计、推荐更新写入 outbox，不在请求内直接调用。
请标出每个函数是否允许访问数据库、是否允许外部 I/O。

这比“帮我重构一下”更有约束力。

HTTP 和 RPC：接口不是函数的远程版本

很多系统从单体拆服务时，会把函数调用直接改成 HTTP 调用。函数名变成 URL，参数变成 JSON，返回值变成 response。

这通常会留下隐患。

本地函数调用失败，抛异常就行。远程调用失败，原因可能是超时、连接失败、对方返回 500、网关返回 502、请求发出但响应丢失、对方成功处理但调用方没收到结果。它不是函数调用的远程版本，它是不可靠网络上的协议交互。

一个看起来普通的 HTTP 调用：

def charge(order_id: str, amount: int):
    response = requests.post(
        "https://payment.example.com/charge",
        json={"order_id": order_id, "amount": amount},
    )
    response.raise_for_status()
    return response.json()

如果 AI 只写到这里，基本还停留在 demo。真实系统至少要问：

• 超时是多少？
• 失败后能不能重试？
• 重试会不会重复扣款？
• 有没有幂等键？
• 错误码如何分类？
• 调用方是否能接受未知状态？
• 对方接口升级时如何兼容？
• 请求和响应是否记录 trace id？

支付这类接口必须有幂等键：

def charge(order_id: str, amount: int):
    idempotency_key = f"charge:{order_id}"
    response = http_client.post(
        "/charge",
        json={"order_id": order_id, "amount": amount},
        headers={"Idempotency-Key": idempotency_key},
        timeout=3,
    )
    return parse_payment_response(response)

这里的关键不是代码写法，而是协议语义。调用方和服务方都要承认：同一个 Idempotency-Key 的重复请求，只能产生一次扣款效果。

RPC 和 gRPC 也一样。它们有更强的 schema，更适合内部高频调用，但仍然绕不开超时、重试、幂等、限流、熔断、版本兼容。

让 AI 设计远程接口时，可以直接要求它输出契约：

请为这个远程调用设计接口契约：
1. 请求字段、响应字段、必填字段和可选字段。
2. 错误码分类：可重试、不可重试、未知状态。
3. 超时设置和重试策略。
4. 幂等键设计。
5. 版本兼容策略，新增字段和废弃字段怎么处理。
6. 日志和 trace 字段。
7. 调用方在超时后如何查询最终状态。

如果 AI 只给 URL 和 JSON 字段，还不够。远程接口的重点是失败语义。

消息队列：解耦不是丢出去就完了

消息队列经常被用来“异步化”。用户下单后发短信，上传文件后做解析，创建报告后后台生成，调用方把消息丢出去就返回。

这听起来很美，但消息队列不是垃圾桶。它只是把同步问题变成了异步问题。

一个最小消息体可能是这样：

{
  "event_id": "6f01d3b8-5d1d-4df2-a4ce-9bb21a8b22b2",
  "event_type": "order.created",
  "occurred_at": "2026-07-03T10:00:00+08:00",
  "payload": {
    "order_id": "order_123",
    "user_id": "user_456"
  }
}

这里最重要的字段不是 payload，而是 event_id。消费者需要靠它做幂等。

消息队列里常见问题包括：

• 消息可能重复投递。
• 消费者处理成功后，提交 offset 或 ack 失败。
• 消费者处理失败后，消息被反复重试。
• 消息顺序只在某个 key 或分区内成立。
• 队列堆积时，延迟越来越高。
• 消费者版本升级后，老消息 schema 不兼容。
• 死信队列没人看，失败变成静默积压。

AI 写消费者时，常常只写：

def consume(message):
    data = json.loads(message.body)
    send_email(data["email"], data["template"])
    message.ack()

这段代码缺少幂等，也没有失败分类。更像样的消费者应该先建立处理记录：

def consume(message):
    event = parse_event(message.body)

    if processed_events.exists(event.event_id):
        message.ack()
        return

    try:
        send_email(event.payload["email"], event.payload["template"])
        processed_events.mark_done(event.event_id)
        message.ack()
    except TemporaryError:
        message.retry_later()
    except PermanentError as exc:
        dead_letters.add(event, reason=str(exc))
        message.ack()

真实实现还要考虑事务边界：发送邮件和写处理记录不能天然放在同一个事务里，所以最好让外部动作本身也支持幂等，或者用 outbox/inbox 模式把状态保存下来。

让 AI 写消息队列逻辑时，不要只说“失败重试”。要写清楚：

消费者必须按 event_id 幂等。
同一 order_id 的消息需要按顺序处理。
临时错误最多重试 5 次，指数退避。
永久错误进入死信队列，并记录可人工排查的信息。
消费成功、重复跳过、重试、死信都要打指标。

这些约束会直接改变代码结构。

事件流：不是所有人都要改调用方

消息队列通常偏任务处理，事件流更像事实发布。订单创建、支付完成、用户资料更新、库存变化，这些都可以作为事件被发布出去，多个消费者各取所需。

事件流的好处是扩展性。今天只有一个统计消费者，明天可以加推荐消费者、风控消费者、审计消费者，不一定要改订单服务。

但事件流需要更强的 schema 纪律。

事件应该描述已经发生的事实，而不是命令。order.created 是事实，send_email 是命令。事实可以被多个消费者理解，命令通常只服务某个消费者。

一个事件 schema 可以这样设计：

{
  "event_id": "uuid",
  "event_type": "order.created",
  "schema_version": 2,
  "aggregate_type": "order",
  "aggregate_id": "order_123",
  "occurred_at": "2026-07-03T10:00:00+08:00",
  "producer": "order-service",
  "payload": {
    "user_id": "user_456",
    "total_amount": 12900,
    "currency": "CNY"
  }
}

这里有几个字段很硬：

• schema_version 用来处理格式演进。
• aggregate_id 用来做同一对象的顺序和聚合。
• producer 方便追踪来源。
• occurred_at 是业务发生时间，不一定等于消息写入时间。

AI 很容易把事件流写成“把当前对象 dump 成 JSON”。这样会泄漏内部字段，也会让消费者绑定生产者的数据库结构。生产者一改表，消费者全跟着坏。

事件 payload 应该是对外稳定的事实模型，而不是内部 ORM 对象。

共享数据库：短期最快，长期最粘

共享数据库是很多系统早期最常见的“通信方式”。A 模块写一张表，B 模块直接读；B 模块需要改状态，也直接 update。

它快，成本低，不需要接口，不需要队列，不需要服务治理。小系统里完全可以用。

问题在于它会绕过业务边界。

假设订单模块有一张表：

orders(id, user_id, status, total_amount, created_at, paid_at)

统计模块直接查它，问题不大。客服模块直接改 status，就危险了。因为订单状态迁移规则可能在订单模块代码里：未支付不能发货，已取消不能支付，退款中不能重复退款。外部模块直接 update 表，就绕过了规则。

共享数据库适合什么？

• 报表和只读分析。
• 小系统内部模块。
• 还没有拆服务的单体。
• 通过只读视图或汇总表对外提供数据。

不适合什么？

• 外部模块直接修改核心业务表。
• 多个服务同时拥有同一张表的写权限。
• 把数据库表结构当成公共 API。
• 靠触发器和隐式更新维持业务规则。

如果必须共享数据库，也可以把伤害降下来：

create view order_report_view as
select id, user_id, status, total_amount, created_at, paid_at
from orders;

只开放视图或只读账号，避免别的模块直接依赖全部表结构。更进一步，可以做专门的读模型或同步表，让业务写模型和分析读模型分开。

让 AI 处理共享数据库时，可以明确要求：

其他模块只能读取 order_report_view，不能直接写 orders 表。
订单状态修改必须通过 order_service.change_status()。
请检查代码里是否存在跨模块直接 update 核心表的行为。

这类约束比“保持低耦合”更有用。

选通信方式时，先画失败路径

很多设计图只画成功路径：

用户请求 -> 订单服务 -> 支付服务 -> 通知服务 -> 完成

成功路径当然要画，但真正决定通信方式的是失败路径。

比如订单创建后需要支付和通知：

• 支付服务超时，订单应该是什么状态？
• 支付请求发出后网络断了，能不能确认是否扣款？
• 通知失败是否影响订单创建？
• 通知重复发送能不能接受？
• 支付成功事件丢了怎么办？
• 订单服务重启后，未完成流程怎么恢复？

如果通知失败不能影响订单，就不要同步卡在主链路里。它适合 outbox + 队列。如果支付超时后状态未知，就需要支付查询接口和补偿任务。如果支付请求可能重复，就必须有幂等键。

一个更接近真实系统的流程可能是：

HTTP 请求
  -> 订单服务创建 pending 订单
  -> 调用支付服务，携带幂等键
  -> 支付成功后，订单进入 paid
  -> 写 outbox: order.paid
  -> 通知消费者异步发送消息
  -> 统计消费者异步更新报表

失败路径也要画：

支付超时
  -> 订单标记为 payment_unknown
  -> 后台任务按 order_id 查询支付最终状态
  -> 查到成功则转 paid
  -> 查到失败则转 payment_failed
  -> 长时间未知则进入人工处理队列

这才是通信方式选择背后的工程逻辑。

版本兼容是模块通信的基本礼貌

模块之间一旦通过接口、消息、事件或数据库视图通信，就会有版本问题。

AI 写代码时，常常假设两边同时升级。这在线上系统里通常不成立。服务可能滚动发布，消费者可能落后，移动端可能几周不升级，队列里可能还有旧消息。

几个基本原则很实用：

• 新增字段尽量可选。
• 不要随便改字段含义。
• 不要直接删除消费者仍在使用的字段。
• 枚举值新增时，旧消费者要能处理未知值。
• 消息 schema 要带版本。
• 读旧版本，写新版本时要有迁移窗口。
• 接口返回字段不应该暴露内部表结构。

比如订单状态新增 payment_unknown。如果旧客户端只认识 pending/paid/cancelled，前端可能直接崩。更好的做法是接口同时返回稳定的展示状态：

{
  "status": "payment_unknown",
  "display_status": "processing",
  "can_cancel": false
}

这样旧客户端至少可以按 display_status 展示“处理中”，而不是把未知枚举当错误。

让 AI 改接口时，可以加一句：

请按滚动发布考虑兼容性：新老服务会同时存在，队列里会有旧消息，旧客户端可能不认识新增枚举。

这句话能挡住很多冒进改法。

可观测性也要跟着通信方式走

模块通信一多，排障就不再是看一个日志文件。

同步调用要有 trace id，能看到请求跨过哪些服务、每一跳耗时多少、在哪里失败。消息队列要有 event id、消费延迟、重试次数、死信数量。事件流要有 offset、lag、schema 版本。共享数据库要能看到慢查询、锁等待和跨模块写入来源。

不同通信方式，至少要留下这些观测点：

通信方式	关键观测点
HTTP / RPC	请求量、延迟、错误率、超时率、重试次数、trace id
消息队列	生产量、消费量、堆积量、消费延迟、重试、死信
事件流	offset、consumer lag、schema version、重放进度
共享数据库	慢查询、锁等待、连接数、调用来源、表级写入
文件交换	文件大小、上传下载耗时、校验结果、清理状态

AI 生成的代码，如果只有业务逻辑，没有这些观测点，出了问题就只能猜。让它补可观测性时，不要泛泛说“加日志”，而要把字段列出来：

每次远程调用日志必须包含：
- trace_id
- caller
- callee
- operation
- request_id 或 event_id
- timeout_ms
- elapsed_ms
- result: success/retryable_error/permanent_error/timeout

结构化日志比一句“调用失败”有用得多。

一份给 AI 的模块通信提示词

写涉及多模块协作的功能时，可以先让 AI 不写代码，只做通信设计：

先不要写代码。

请为这个功能设计模块通信方式：
1. 列出参与模块和每个模块拥有的数据。
2. 区分同步核心路径、异步副作用、只读查询和批处理任务。
3. 为每条交互选择通信方式：函数调用、HTTP/RPC、消息队列、事件流、共享数据库、文件交换。
4. 说明每条交互是否需要等待结果，是否在同一个事务里，失败如何处理。
5. 给出接口或消息 schema，包括版本字段、幂等键、trace 字段。
6. 分析超时、重试、重复调用、乱序、消息丢失、消费者落后的处理方式。
7. 列出不能跨模块直接访问的数据表或内部字段。
8. 给出观测指标和日志字段。
9. 最后再写代码实现计划。

如果 AI 的回答里没有失败路径、幂等、版本兼容和观测字段，说明它还没有真正设计模块通信，只是在生成调用代码。

写代码之前，先问几句话

模块通信方式选得好，系统会清爽很多；选错了，后面每个需求都会带着历史包袱。

动手前可以问自己：

• 这次交互必须同步等结果吗？
• 失败会不会影响主流程？
• 重试会不会造成重复副作用？
• 有没有幂等键？
• 接口或消息 schema 是否稳定？
• 两边能不能独立发布？
• 消费者落后或下线时会怎样？
• 数据表是不是被多个模块共同写入？
• 出问题时能不能按 trace id 或 event id 找到完整路径？

这些问题都不新潮，但它们是真正的工程边界。

AI 可以很快写出 HTTP client、队列消费者、RPC handler、数据库查询和事件发布器。程序员要做的，是先决定这些模块到底该怎么说话，什么时候应该沉默，什么时候应该等待，什么时候应该把事情记下来交给后台慢慢做。

代码会变，边界会留下。

微信公众号

欢迎关注「字与码」

如果这篇文章对你有用，也欢迎在微信里继续关注后续更新。

X / Twitter

关注 @ax2_zicode

更即时的技术观察、新文章提醒和一些短想法会发在 X 上。

X 打开 X 主页 @ax2_zicode