Publisher 与 SyncInTxn

包：publish

事件追加之后，Repository 把它们交给 Publisher。go-sequent 有一个默认模式和一个可选的附加模式。

SyncInTxn（默认）

publish.SyncInTxn 在写事务里同步更新读模型：对每个事件，按顺序先跑注册到该类型的 Projector，再跑 Workflow。任何处理器返回错误都会中止事务，什么都不会落库。

type Handler interface {
    Handles() []string
    Handle(ctx context.Context, e event.Event) error
}

通过门面注册处理器：

app.RegisterProjector(balanceProjector) // 参与读模型重建
app.RegisterWorkflow(notifyWorkflow)    // 不参与读模型重建

它是默认模式的原因：账本余额必须在命令返回时就是正确的。Projector 与追加在同一个事务里跑，就不存在「事件已存在而余额还没更新」的窗口。也就是提交返回后立刻能读到自己刚写的数据，见 名词释义。账本读模型必须使用 SyncInTxn。

代价是 Projector 的工作量落在写路径上。账本 Projector 要保持轻量且确定，把重的或对外的工作放到别处。

顺序

同一次 Publish 中，事件按聚合加载顺序到达，每个聚合内部再按序号。对每个事件，Projector 先于 Workflow 运行；同一事件的多个处理器按注册顺序运行。

失败

Projector 错误会被包装为 publish: projector for <EventType>: ...，Workflow 错误会被包装为 publish: workflow for <EventType>: ...。任何一个都会回滚整个 Execute。

PublishProjectorsOnly

PublishProjectorsOnly 只运行 Projector，跳过 Workflow。读模型重建流水线使用它，这样回放历史时不会重复发送通知，也不会重复追加命令。

WithOutbox

需要异步扇出时，用 publish.WithOutbox 包装 Publisher：

ob := outbox.NewMemory() // 或 outbox.NewPostgres(pool)
pub := &publish.WithOutbox{
    Inner:  app.Publisher,
    Writer: ob,
    Topic:  "events",
    // EventTypes: map[string]bool{"bank.MoneyDeposited": true}, // 为空表示全部事件
}
app.Repo.Publisher = pub

Publish 先运行内部 SyncInTxn 的处理器，再为每个匹配的事件以 JSON 入队一条 Outbox 消息。使用 Postgres writer 时，消息行在写事务里插入，所以事件、读模型和 Outbox 行要么一起提交，要么一起回滚。

传输方式是一张由轮询器消费的 Postgres 表。框架里没有 LISTEN/NOTIFY、Kafka 或其他消息中间件。

适合用 Outbox 的场景：

• 外部系统需要在提交之后看到事件。
• 跨服务集成。
• 对写延迟预算来说太重、且不是账本读模型的更新。

不要把它用于账本余额。

消费 Outbox

poller := &outbox.Poller{
    Store:      ob, // outbox.Store：Claim、Ack、Nack、Depth
    Handler:    func(ctx context.Context, msg outbox.Message) error { /* 投递 */ return nil },
    BatchSize:  10,
    Interval:   time.Second,
    LockFor:    30 * time.Second,
    RetryAfter: 5 * time.Second,
}
err := poller.Run(ctx) // 或 RunOnce(ctx)

Claim 锁定一批消息，Handler 逐条投递，成功后 poller 调用 Ack，失败则带着重试延迟和错误文本调用 Nack。投递语义是至少一次，消费端必须保证同一业务键重复提交只生效一次。Message 包含 Topic、Payload、TenantID、AggregateID、EventType、Attempts 和时间戳。

CLI 提供轮询命令：

go run ./cmd/go-sequent outbox -dsn "$GO_SEQUENT_PG_DSN" -once
go run ./cmd/go-sequent outbox -dsn "$DSN" -interval 1s -batch 10

自带 CLI 的 Handler 只会打印每条消息，它不知道你的投递目标。要真正投递，请在自己的服务里运行 outbox.Poller，并让 Handler 调用目标系统。

Store.Depth 返回待处理数量，可传给 obs.Config.OutboxDepth 以得到 sequent.outbox.depth 指标。见 OpenTelemetry。

如何选择

相关

• Projector
• Workflow
• 读模型重建

文档随 go-sequent 源码更新；若与 API 不一致，以源码为准。