三个词不是同义词
“发个消息”“发个事件”“发个命令”在日常讨论里经常混着说，落到代码里却变成同一个 Publish(topic, payload)。站内 消息驱动架构
已经讲过事件驱动的整体结构，本节只收紧一个更小的边界：这三个词描述的到底是三种东西，还是同一种东西的三个名字。
先给结论。事件是已经发生的事实，时态是过去，发布者不关心谁消费、也不接受业务拒绝；命令是希望发生的意图，时态是祈使，必须有一个明确的接收者，并且可以被拒绝；消息不是前两者的同级概念，它是承载事件或命令的传输信封，负责 ID、类型、时间、关联 ID 和负载。把消息当成第三种语义，就会得到“发了消息但不知道对方该不该拒绝”的代码。
订单域里的三种东西
在 arch-lab 的下单流程里，三种东西可以同时出现：
OrderPlaced是事件：订单已经创建，事实无法被拒绝。它可以有投影、通知、积分三个订阅者，也可以暂时一个都没有。ReserveInventory是命令：编排器要求库存服务预留指定 SKU，库存服务可以返回OUT_OF_STOCK。这个拒绝必须回到编排器，否则订单会停在错误状态。Envelope是消息：它把上面两者序列化后送进 Kafka、NATS 或进程内总线，自身不解释业务含义，只保证“这是什么类型、属于哪次请求”。
| 维度 | 事件 Event | 命令 Command | 消息 Message |
|---|---|---|---|
| 语义 | 已发生的事实 | 希望发生的意图 | 传输信封 |
| 时态 | 过去 | 祈使 | 无 |
| 接收者 | 0 到 N 个 | 恰好 1 个 | 由路由决定 |
| 业务拒绝 | 不存在 | 可以拒绝 | 只反映投递结果 |
| 重试责任 | 各消费者幂等 | 发布者决定 | 传输层重投 |
| arch-lab 例子 | OrderPlaced | ReserveInventory | Envelope |
这张表最重要的两行是“接收者”和“业务拒绝”。事件的发布者不应该知道订阅者是谁，所以新增一个订阅者不需要改发布方；命令的发布者必须知道唯一接收者，否则“库存不足”这个拒绝没有地方返回。很多“事件驱动”事故，本质是把这两行合并了。
误用之后会发生什么
第一种误用是把事件当命令。订单服务发布 OrderPlaced，支付消费者订阅后直接扣款。发布方以为自己在广播事实，实际上触发了一个有外部副作用的动作；新增一个消费者就可能多扣一次款，而且发布方无法知道扣款失败。事件里如果还塞了 please_charge: true 这类字段，等于用事实的壳包装命令。
第二种误用是把命令当广播。ReserveInventory 被两个消费者订阅，各自扣一次库存；或者用 fire-and-forget 发布，库存不足的拒绝被吞掉，编排器只能等超时。命令的语义要求“恰好一个处理者”和“结果回到发布者”，这两个约束必须在类型或总线层强制，而不是靠文档。
第三种误用是消息不带 Kind。消费者拿到一个 JSON，不知道应该校验成事件还是命令，也不知道失败时该返回 4xx、重试还是记录后忽略。路由键只说明“去哪”，不说明“该做什么”。
第四种误用是事件里写指令。OrderPlaced 的负载里带 next_action: "reserve_inventory"，消费者被迫按发布者的时间表行动，订阅者之间产生隐式顺序依赖。事件应该描述已经发生的事实，编排决策属于命令发布者。
把语义写进类型：Envelope 与总线
先定义信封。Kind 只有 event 和 command 两个合法值；命令额外要求 CorrelationID，因为没有关联 ID 就无法把拒绝、重试和审计串回同一次业务请求。Validate 在进入总线前拦截非法信封，避免错误在消费者侧才暴露。
// file: internal/messaging/envelope.go
// tier: L1
// verify: go test -race -v ./internal/messaging/
package messaging
import (
"errors"
"fmt"
"time"
)
type Kind string
const (
KindEvent Kind = "event"
KindCommand Kind = "command"
)
var (
ErrInvalidEnvelope = errors.New("invalid envelope")
ErrNoHandler = errors.New("no handler")
ErrHandlerExists = errors.New("handler already registered")
)
// Envelope 是传输层唯一认识的类型：Kind 决定语义，Name 决定路由，
// Payload 只承载领域序列化结果，不参与路由。
type Envelope struct {
ID string
Kind Kind
Name string
OccurredAt time.Time
CorrelationID string
Payload []byte
}
func (e Envelope) Validate() error {
switch {
case e.ID == "":
return fmt.Errorf("%w: missing id", ErrInvalidEnvelope)
case e.Kind != KindEvent && e.Kind != KindCommand:
return fmt.Errorf("%w: unknown kind %q", ErrInvalidEnvelope, e.Kind)
case e.Name == "":
return fmt.Errorf("%w: missing name", ErrInvalidEnvelope)
case e.OccurredAt.IsZero():
return fmt.Errorf("%w: missing occurred_at", ErrInvalidEnvelope)
case e.Kind == KindCommand && e.CorrelationID == "":
return fmt.Errorf("%w: command requires correlation_id", ErrInvalidEnvelope)
}
return nil
}
func NewEvent(id, name string, at time.Time, payload []byte) (Envelope, error) {
e := Envelope{ID: id, Kind: KindEvent, Name: name, OccurredAt: at, Payload: payload}
return e, e.Validate()
}
func NewCommand(id, name, correlationID string, at time.Time, payload []byte) (Envelope, error) {
e := Envelope{
ID: id, Kind: KindCommand, Name: name, CorrelationID: correlationID,
OccurredAt: at, Payload: payload,
}
return e, e.Validate()
}
总线把两种语义拆成两套注册接口。Subscribe 用于事件，允许同名多注册，也允许零订阅；RegisterCommand 用于命令，同名第二次注册直接返回 ErrHandlerExists。Publish 根据 Kind 分流：事件遍历所有订阅者并用 errors.Join 收集错误，某个订阅者失败不影响其他订阅者；命令只找一个处理者，找不到返回 ErrNoHandler，处理者的业务错误原样回到发布者。
// file: internal/messaging/bus.go
// tier: L1
// verify: go test -race -v ./internal/messaging/
package messaging
import (
"context"
"errors"
"fmt"
"sync"
)
type Handler func(context.Context, Envelope) error
type Publisher interface {
Publish(ctx context.Context, e Envelope) error
}
type Subscriber interface {
Subscribe(name string, h Handler)
}
type Bus struct {
mu sync.RWMutex
events map[string][]Handler
commands map[string]Handler
}
func NewBus() *Bus {
return &Bus{events: make(map[string][]Handler), commands: make(map[string]Handler)}
}
// Subscribe 只用于事件：同一事件可以有零个、一个或多个消费者。
func (b *Bus) Subscribe(name string, h Handler) {
b.mu.Lock()
defer b.mu.Unlock()
b.events[name] = append(b.events[name], h)
}
// RegisterCommand 只用于命令：同一命令名只允许一个处理者。
func (b *Bus) RegisterCommand(name string, h Handler) error {
b.mu.Lock()
defer b.mu.Unlock()
if _, ok := b.commands[name]; ok {
return fmt.Errorf("%w: %s", ErrHandlerExists, name)
}
b.commands[name] = h
return nil
}
func (b *Bus) Publish(ctx context.Context, e Envelope) error {
if err := e.Validate(); err != nil {
return err
}
b.mu.RLock()
switch e.Kind {
case KindEvent:
handlers := append([]Handler(nil), b.events[e.Name]...)
b.mu.RUnlock()
// 事件是已经发生的事实：发布者不等待业务结果，某个订阅者失败
// 也不阻止其他订阅者收到它。
return fanOut(ctx, handlers, e)
case KindCommand:
h, ok := b.commands[e.Name]
b.mu.RUnlock()
if !ok {
return fmt.Errorf("%w: %s", ErrNoHandler, e.Name)
}
// 命令有明确的接收者：拒绝和失败必须回到发布者。
return h(ctx, e)
}
b.mu.RUnlock()
return fmt.Errorf("%w: unknown kind %q", ErrInvalidEnvelope, e.Kind)
}
func fanOut(ctx context.Context, handlers []Handler, e Envelope) error {
var errs []error
for _, h := range handlers {
if err := h(ctx, e); err != nil {
errs = append(errs, err)
}
}
return errors.Join(errs...)
}
这里有一个容易忽略的取舍：事件发布返回 error 并不代表“事件被拒绝”，它只表示“至少一个订阅者投递失败”。调用方可以记录、告警或重投，但不能因此回滚已经发生的事实。命令返回的 error 则是业务结果，调用方必须处理。同一个 Publish 签名承载两种错误语义，所以类型里的 Kind 不是装饰，而是调用方解释返回值的唯一依据。
表驱动测试与真实输出
测试先覆盖信封校验的五种输入，再分别验证事件扇出、订阅者失败隔离、零订阅、命令单处理者、命令无处理者和命令拒绝回传。所有计数器用 atomic，因为测试要在 -race 下跑。
// file: internal/messaging/bus_test.go
// tier: L1
// verify: go test -race -v ./internal/messaging/
package messaging
import (
"context"
"errors"
"sync/atomic"
"testing"
"time"
)
var (
ErrOutOfStock = errors.New("out of stock")
ErrProjection = errors.New("projection failed")
)
func at() time.Time { return time.Date(2026, 10, 9, 0, 0, 0, 0, time.UTC) }
func TestEnvelopeValidation(t *testing.T) {
cases := []struct {
name string
env Envelope
want error
}{
{"event", Envelope{ID: "e-1", Kind: KindEvent, Name: "OrderPlaced", OccurredAt: at()}, nil},
{"command", Envelope{ID: "c-1", Kind: KindCommand, Name: "ReserveInventory", CorrelationID: "req-1", OccurredAt: at()}, nil},
{"missing id", Envelope{Kind: KindEvent, Name: "OrderPlaced", OccurredAt: at()}, ErrInvalidEnvelope},
{"unknown kind", Envelope{ID: "x", Kind: "query", Name: "GetOrder", OccurredAt: at()}, ErrInvalidEnvelope},
{"command without correlation", Envelope{ID: "c-2", Kind: KindCommand, Name: "ReserveInventory", OccurredAt: at()}, ErrInvalidEnvelope},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := tc.env.Validate()
if tc.want == nil && err != nil {
t.Fatalf("err = %v", err)
}
if tc.want != nil && !errors.Is(err, tc.want) {
t.Fatalf("err = %v, want %v", err, tc.want)
}
})
}
}
func TestEventFansOutToEverySubscriber(t *testing.T) {
bus := NewBus()
var calls atomic.Int32
for i := 0; i < 2; i++ {
bus.Subscribe("OrderPlaced", func(context.Context, Envelope) error {
calls.Add(1)
return nil
})
}
ev, err := NewEvent("e-1", "OrderPlaced", at(), []byte(`{"order_id":"o-1"}`))
if err != nil {
t.Fatal(err)
}
if err := bus.Publish(context.Background(), ev); err != nil {
t.Fatal(err)
}
if calls.Load() != 2 {
t.Fatalf("calls = %d, want 2", calls.Load())
}
}
func TestEventContinuesAfterSubscriberError(t *testing.T) {
bus := NewBus()
var second atomic.Bool
bus.Subscribe("OrderPlaced", func(context.Context, Envelope) error { return ErrProjection })
bus.Subscribe("OrderPlaced", func(context.Context, Envelope) error {
second.Store(true)
return nil
})
ev, _ := NewEvent("e-1", "OrderPlaced", at(), nil)
if err := bus.Publish(context.Background(), ev); !errors.Is(err, ErrProjection) {
t.Fatalf("err = %v, want ErrProjection", err)
}
if !second.Load() {
t.Fatal("second subscriber was skipped")
}
}
func TestEventWithNoSubscriberIsAccepted(t *testing.T) {
ev, _ := NewEvent("e-1", "OrderPlaced", at(), nil)
if err := NewBus().Publish(context.Background(), ev); err != nil {
t.Fatalf("event with no subscriber: %v", err)
}
}
func TestCommandAllowsExactlyOneHandler(t *testing.T) {
bus := NewBus()
h := func(context.Context, Envelope) error { return nil }
if err := bus.RegisterCommand("ReserveInventory", h); err != nil {
t.Fatal(err)
}
if err := bus.RegisterCommand("ReserveInventory", h); !errors.Is(err, ErrHandlerExists) {
t.Fatalf("second register err = %v, want ErrHandlerExists", err)
}
cmd, _ := NewCommand("c-1", "ReserveInventory", "req-1", at(), nil)
if err := bus.Publish(context.Background(), cmd); err != nil {
t.Fatalf("publish command: %v", err)
}
}
func TestCommandWithoutHandlerIsRejected(t *testing.T) {
cmd, _ := NewCommand("c-1", "ReserveInventory", "req-1", at(), nil)
if err := NewBus().Publish(context.Background(), cmd); !errors.Is(err, ErrNoHandler) {
t.Fatalf("err = %v, want ErrNoHandler", err)
}
}
func TestCommandRejectionReachesPublisher(t *testing.T) {
bus := NewBus()
_ = bus.RegisterCommand("ReserveInventory", func(context.Context, Envelope) error {
return ErrOutOfStock
})
cmd, _ := NewCommand("c-1", "ReserveInventory", "req-1", at(), nil)
if err := bus.Publish(context.Background(), cmd); !errors.Is(err, ErrOutOfStock) {
t.Fatalf("err = %v, want ErrOutOfStock", err)
}
}
以下输出来自本机 macOS（darwin/arm64，Go 1.26.0），在独立模块 .local/checks/ch07-be-arch-base-07-2-event-command-message 中执行。gofmt -l . 无输出，go vet ./... 通过。
$ go test -race -v ./internal/messaging/
--- PASS: TestEnvelopeValidation (0.00s)
--- PASS: TestEnvelopeValidation/event (0.00s)
--- PASS: TestEnvelopeValidation/command (0.00s)
--- PASS: TestEnvelopeValidation/missing_id (0.00s)
--- PASS: TestEnvelopeValidation/unknown_kind (0.00s)
--- PASS: TestEnvelopeValidation/command_without_correlation (0.00s)
--- PASS: TestEventFansOutToEverySubscriber (0.00s)
--- PASS: TestEventContinuesAfterSubscriberError (0.00s)
--- PASS: TestEventWithNoSubscriberIsAccepted (0.00s)
--- PASS: TestCommandAllowsExactlyOneHandler (0.00s)
--- PASS: TestCommandWithoutHandlerIsRejected (0.00s)
--- PASS: TestCommandRejectionReachesPublisher (0.00s)
PASS
ok archlab/internal/messaging 1.332s
与已有事件驱动文章的分工
站内 消息驱动架构 回答的是“什么时候该用事件驱动、事件如何落库、如何避免双写”，属于架构风格与一致性层面。本节回答更前置的问题：“这条消息到底该被当成事实还是意图”。前者决定系统形状，后者决定单个接口的契约；如果语义层没定清，事件驱动的基础设施越完整，误用传播得越快。消息可靠投递与幂等的细节放在 7.3 和卷二展开，这里不重复。
常见错误与边界
第一，用 topic 名称代替语义。order-events 里既放 OrderPlaced 又放 CancelOrder，消费者只能靠 payload 猜。正确做法是让 Kind 成为信封的必填字段，topic 只负责粗粒度隔离。
第二，认为命令必须同步返回。命令可以有异步执行路径，但“接受/拒绝”的确认必须回到发布者；如果连接受与否都不返回，它就已经退化成事件，不该叫命令。
第三，把事件消费者的失败当成发布失败。事件已经发生，发布方无法撤销；消费者失败应由重试、死信和补偿处理，不能反向阻塞事实的产生。
第四，为了“解耦”把所有命令都改成事件。命令的拒绝能力是业务需要的，库存不足、支付失败、状态非法都必须有明确回传路径；把这些改成事件，只会把失败检测推迟到对账。
小结
事件、命令、消息的差异不在传输介质，而在契约：事件是零到多个消费者共享的既成事实，命令是恰好一个接收者且可拒绝的意图，消息是承载两者的信封。arch-lab 用 Kind 校验、Subscribe 与 RegisterCommand 两套注册接口、以及“事件聚合错误、命令原样回传”的发布逻辑，把这三条契约写成了可执行代码。下一节继续往契约内部走：当同一条命令或同一个事件被重复投递时，幂等责任究竟该由 API、领域还是消息层承担。
阅读导航：上一节：7.1 同步调用链的隐性耦合 · 下一节：7.3 幂等责任归属：API、领域、消息层的契约 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。