02 - 核心概念

在深入 Watermill 代码之前，必须先理解其核心概念。以下关系图展示了各组件的结构与交互：

graph TB
    subgraph "Watermill Core"
        MSG[Message<br/>UUID + Payload + Metadata + Ack/Nack]
        PUB[Publisher<br/>Publish(topic, messages...)]
        SUB[Subscriber<br/>Subscribe(ctx, topic) -> chan]
    end

    subgraph "Router & Middleware"
        R[Router<br/>Handler 注册 + Topic 路由]
        MW1[Recoverer]
        MW2[Retry]
        MW3[Timeout]
        MW4[Throttle]
        MW1 --> MW2 --> MW3 --> MW4 --> H
    end

    subgraph "CQRS Layer"
        CB[CommandBus]
        EB[EventBus]
        CM[CommandHandler]
        EH[EventHandler]
    end

    PUB -->|Publish| MSG -->|Subscribe| SUB
    SUB -->|chan *Message| R
    R -->|Middleware Chain| H[Handler<br/>func(*Message) -> []*Message, error]
    CB --> PUB
    EB --> PUB

Message（消息）

Message 是 Watermill 的核心数据结构，定义在 message 包中：

字段	类型	说明
UUID	string	全局唯一标识，由 watermill.NewUUID() 生成
Payload	[]byte	消息体，任意序列化后的字节流
Metadata	Metadata (map[string]string)	键值对元数据，用于链路追踪、分区键等
ack	channel	ACK 信号通道
nack	channel	NACK 信号通道

一个典型的创建方式（参见 basic/01-pubsub/main.go:38）：

msg := message.NewMessage(watermill.NewUUID(), payload)

Publisher（发布者）

Publisher 是发布消息的接口，只有唯一方法：

type Publisher interface {
    Publish(topic string, messages ...*Message) error
}

同一个 topic 可以有多个 Publisher，Watermill 保证消息至少投递一次。在 basic/01-pubsub/main.go:39 中可以看到最简单的用法：

pubSub.Publish("hello", msg)

Subscriber（订阅者）

Subscriber 消费消息，返回 Go channel：

type Subscriber interface {
    Subscribe(ctx context.Context, topic string) (<-chan *Message, error)
}

返回值 <-chan *Message 在 GoChannel 实现中（默认第 line 22）是带缓冲的 channel，缓冲大小通过 Config.OutputChannelBuffer 配置。消息消费完毕后需要显式调用 msg.Ack() 确认。

Router（路由器）

Router 是 Watermill 最核心的组件，负责将 Subscriber 收到的消息路由到对应的 Handler 函数。Router 连接 "谁发布到哪" 和 "谁处理什么" 的映射关系（参见 basic/02-router/main.go:29）。

Handler 签名固定为：

func(msg *message.Message) ([]*Message, error)

Handler 可以返回新的消息列表（通过 Router 发布到输出 topic），也可以返回 nil 表示不产生新消息。AddNoPublisherHandler 则用于不需要输出 topic 的消费场景。

Middleware（中间件）

中间件采用责任链模式，在消息到达 Handler 前后执行。Watermill 内置了 Recoverer（panic 恢复）、Retry（失败重试）、Timeout（超时控制）、Throttle（限流）等。中间件按 router.AddMiddleware() 的注册顺序依次执行。

CQRS（命令查询职责分离）

Watermill 的 CQRS 组件（components/cqrs 包）在 Pub/Sub 基础上封装了 CommandBus 和 EventBus，提供类型安全的命令/事件处理模式。Command 表示意图（"创建订单"），Event 表示已发生的事实（"订单已创建"）。