分布式系统在数据库写入成功但事件发布失败时可能会出问题。

在本教程中，@pliutau 将教你如何在 Go 和 PostgreSQL 中实现 Outbox 模式。

你还将学习如何构建一个 relay 服务、使用 outbox 表，以及在事件驱动系统中处理至少一次交付。

https://freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/…

如何在 Go 和 PostgreSQL 中实现 Outbox 模式

来源：https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/ 如何在 Go 和 PostgreSQL 中实现 Outbox 模式在事件驱动系统中，处理一个请求时需要做两件事：你需要将数据保存到数据库，并且需要向消息代理发布一个事件，以便其他服务知道有变化发生。

这两个操作看起来很简单，但它们隐藏着一个危险的可靠性问题。如果数据库写入成功但消息代理暂时不可用怎么办？或者你的服务在这两步之间崩溃了怎么办？最终你会处于不一致的状态：你的数据库有了新数据，但系统的其余部分从未得知此事。

Outbox 模式是解决这个问题的成熟方案。在本教程中，你将了解这种模式是什么、为什么有效，以及如何用 Go 搭配 PostgreSQL 和 Google Cloud Pub/Sub 实现它。

先决条件

在阅读本教程之前，你应该熟悉：

• Go 编程语言的基础知识
• SQL 和 PostgreSQL
• 数据库事务的概念
• 对事件驱动或分布式系统有基本了解（有帮助但不是必需的）

目录

1. 问题：两个操作，没有原子性 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-the-problem-two-operations-no-atomicity)
2. Outbox 模式如何工作 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-how-the-outbox-pattern-works)
3. Outbox 表结构 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-the-outbox-table-schema)
4. 消息 Relay (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-the-message-relay)
5. Go 和 PostgreSQL 实现 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-go-and-postgresql-implementation) - 订单服务 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-the-orders-service) - Relay 服务 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-the-relay-service)
6. 为什么消息可能会被投递多次 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-why-messages-can-be-delivered-more-than-once)
7. 替代方案：PostgreSQL 逻辑复制 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-alternative-postgresql-logical-replication)
8. 结论 (https://www.freecodecamp.org/news/how-to-implement-the-outbox-pattern-in-go-and-postgresql/#heading-conclusion)

问题：两个操作，没有原子性

要理解为什么会出现 Outbox 模式，你需要了解分布式系统中的一个核心挑战：跨不同系统的原子性。

在关系型数据库中，事务允许你将多个操作分组，使它们要么全部成功，要么全部失败。如果你在同一个事务中插入一行并更新另一行，那么可以保证两者都会发生——或者两者都不会发生。

当你试图将这种保证扩展到两个不同的系统时，问题就出现了：例如，你的数据库和你的消息代理（如 Kafka、RabbitMQ 或 Pub/Sub）。这些系统不共享事务边界。

下面是没有 Outbox 模式时会发生故障的典型事件驱动流程：

1. 用户下订单。
2. 你的服务将订单保存到数据库 ✅
3. 你的服务向消息代理发布 order.created 事件 ❌（代理已宕机）
4. 订单存在于数据库中，但下游服务从未得知此事。

或者相反的故障：

1. 你的服务先发布事件 ✅
2. 你的服务尝试将订单保存到数据库 ❌（数据库超时）
3. 下游服务收到了一个不存在订单的通知。

任何一种情况都会使你的系统处于不一致状态。这就是 Outbox 模式要解决的核心问题。

以下是不使用 Outbox 模式时的流程：

没有 outbox 的图示Outbox 模式通过将两个操作都保留在数据库内部来解决原子性问题：

1. 将你的业务数据（例如，新订单）保存到数据库中。
2. 在同一个数据库事务中，将事件消息写入一个称为 outbox 表的特殊表。
3. 一个名为 Message Relay 的独立后台进程轮询 outbox 表，并将待处理的消息发布到代理。
4. 一旦代理确认收到，relay 就将消息标记为已处理。

因为步骤 1 和 2 发生在同一个数据库事务中，所以它们是原子的。要么两者都成功，要么两者都不成功。你永远不会出现已保存数据但未排队相应事件的情况——或者为从未保存的数据排队事件。

消息永远不会在你的主应用程序代码中直接发布到代理。相反，数据库充当一个可靠的中转区域。

带 outbox 的图示## Outbox 表结构

outbox 表存储待处理的消息，直到 relay 将其取出。以下是典型的 PostgreSQL 表结构：

CREATE TABLE outbox ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), topic varchar(255) NOT NULL, message jsonb NOT NULL, state varchar(50) NOT NULL DEFAULT 'pending', created_at timestamptz NOT NULL DEFAULT now(), processed_at timestamptz );

让我们逐一了解每一列：

• id：每条消息的唯一标识符。使用 UUID 可以方便地引用特定消息。
• topic：消息代理中的目标主题或队列名称（例如，orders.created）。
• message：事件负载，存储为 JSON。这是你的消费者将接收的数据。
• state：跟踪消息是否已发送。两个主要值是pending（等待发布）和processed（已成功发布）。
• created_at：消息插入的时间。relay 使用它按顺序处理消息。
• processed_at：relay 成功发布消息的时间。

你可能需要根据需求添加额外的列：例如，retry_count 列来跟踪 relay 尝试发送消息的次数，或者 error 列来记录失败原因。

消息 Relay

Message Relay 是一个后台进程（通常是 goroutine、sidecar 或独立服务），它连接 outbox 表和消息代理。

它的职责是：

1. 定期查询 outbox 表中 state = 'pending' 的消息。
2. 将每条消息发布到代理中相应的主题。
3. 一旦代理确认投递，将行的 state 更新为 'processed'。
4. 优雅地处理失败：如果发布失败，保持消息为 'pending'，以便重试。

这种设计为你提供了至少一次投递：即使 relay 崩溃并重启，消息也总是会被发送。代价是消息有时可能会被发送多次（下面会有更多介绍），因此你的消费者应该处理重复消息。

Go 和 PostgreSQL 实现

让我们构建一个具体示例。假设你有一个订单服务。创建新订单时，你需要：

1. 将订单保存到 PostgreSQL 的 orders 表中。
2. 将 order.created 事件发布到 Google Cloud Pub/Sub。

你将使用 pgx (https://github.com/jackc/pgx) 作为 PostgreSQL 驱动程序。

订单服务

关键在于订单插入和 outbox 插入发生在同一个事务中。如果任何地方出错，两者都会回滚。

`` // orders/main.go

package main

import ( “context” “encoding/json” “log” “os”

"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"

)

// Order represents a customer order in our system. type Order struct { ID uuid.UUID json:"id" Product string json:"product" Quantity int json:"quantity" }

// OrderCreatedEvent is the payload published to the message broker. // It contains only the fields that downstream services need to know about. type OrderCreatedEvent struct { OrderID uuid.UUID json:"order_id" Product string json:"product" }

// createOrderInTx saves a new order and its outbox event atomically. // Both operations share the same transaction (tx), so either both succeed // or both are rolled back — ensuring consistency. func createOrderInTx(ctx context.Context, tx pgx.Tx, order Order) error { // Step 1: Insert the business data (the actual order). _, err := tx.Exec(ctx, “INSERT INTO orders (id, product, quantity) VALUES ((1, )2, $3)”, order.ID, order.Product, order.Quantity, ) if err != nil { return err } log.Printf(“Inserted order %s into database”, order.ID)

// Step 2: Serialize the event payload that consumers will receive.
event := OrderCreatedEvent{
	OrderID: order.ID,
	Product: order.Product,
}
msg, err := json.Marshal(event)
if err != nil {
	return err
}

// Step 3: Write the event to the outbox table.
// This does NOT publish to Pub/Sub — it just queues it for the relay.
_, err = tx.Exec(ctx,
	"INSERT INTO outbox (topic, message) VALUES (\(1, \)2)",
	"orders.created", msg,
)
if err != nil {
	return err
}
log.Printf("Inserted outbox event for order %s", order.ID)

return nil

}

func main() { ctx := context.Background()

pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
	log.Fatalf("Unable to connect to database: %v", err)
}
defer pool.Close()

// Begin a transaction that will cover both the order insert
// and the outbox insert.
tx, err := pool.Begin(ctx)
if err != nil {
	log.Fatalf("Unable to begin transaction: %v", err)
}
// If anything fails, the deferred Rollback is a no-op after a successful Commit.
defer tx.Rollback(ctx)

newOrder := Order{
	ID:       uuid.New(),
	Product:  "Super Widget",
	Quantity: 10,
}

if err := createOrderInTx(ctx, tx, newOrder); err != nil {
	log.Fatalf("Failed to create order: %v", err)
}

// Committing the transaction makes both writes permanent simultaneously.
if err := tx.Commit(ctx); err != nil {
	log.Fatalf("Failed to commit transaction: %v", err)
}

log.Println("Successfully created order and queued outbox event.")

} ``

请注意，createOrderInTx 接收的是 pgx.Tx（一个事务）而不是连接池。这是有意为之：它强制调用方负责管理事务边界，使原子性保证明确。

Relay 服务

relay 作为独立的后台进程运行。它轮询 outbox 表，发布消息，并将其标记为已处理。

这里一个关键细节是在 SQL 查询中使用了 FOR UPDATE SKIP LOCKED。这个 PostgreSQL 特性允许你并发运行多个 relay 实例而不会互相干扰。当一个实例锁定一行进行处理时，其他实例会跳过该行并继续处理下一行。

`` // relay/main.go

package main

import ( “context” “log” “time”

"cloud.google.com/go/pubsub"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"

)

// OutboxMessage mirrors the columns we need from the outbox table. type OutboxMessage struct { ID uuid.UUID Topic string Message []byte }

// processOutboxMessages picks up one pending message, publishes it to Pub/Sub, // and marks it as processed — all within a single database transaction. func processOutboxMessages(ctx context.Context, pool *pgxpool.Pool, pubsubClient *pubsub.Client) error { tx, err := pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx)

// Query for the next pending message.
// FOR UPDATE SKIP LOCKED ensures that if multiple relay instances are
// running, they won't try to process the same message simultaneously.
rows, err := tx.Query(ctx, `
	SELECT id, topic, message
	FROM outbox
	WHERE state = 'pending'
	ORDER BY created_at
	LIMIT 1
	FOR UPDATE SKIP LOCKED
`)
if err != nil {
	return err
}
defer rows.Close()

var msg OutboxMessage
if rows.Next() {
	if err := rows.Scan(&msg.ID, &msg.Topic, &msg.Message); err != nil {
		return err
	}
} else {
	// No pending messages — nothing to do.
	return nil
}

log.Printf("Publishing message %s to topic %s", msg.ID, msg.Topic)

// Publish the message to the Pub/Sub topic and wait for confirmation.
result := pubsubClient.Topic(msg.Topic).Publish(ctx, &pubsub.Message{
	Data: msg.Message,
})
if _, err = result.Get(ctx); err != nil {
	// Publishing failed. We return the error here without committing,
	// so the transaction rolls back and the message stays 'pending'.
	// The relay will retry it on the next polling interval.
	return err
}

// Mark the message as processed now that the broker has confirmed receipt.
_, err = tx.Exec(ctx,
	"UPDATE outbox SET state = 'processed', processed_at = now() WHERE id = $1",
	msg.ID,
)
if err != nil {
	return err
}
log.Printf("Marked message %s as processed", msg.ID)

// Commit the transaction: the state update becomes permanent.
return tx.Commit(ctx)

}

func main() { // In production, initialize real connections using environment variables // or a config file. These are left as placeholders for clarity. var ( pool *pgxpool.Pool pubsubClient *pubsub.Client )

// Poll the outbox table every second.
// Adjust the interval based on your latency requirements.
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()

for range ticker.C {
	if err := processOutboxMessages(context.Background(), pool, pubsubClient); err != nil {
		log.Printf("Error processing outbox: %v", err)
	}
}

} ``

轮询间隔（本例中为 1 秒）控制着事件写入 outbox 到发布到代理之间的最大延迟。对于大多数用例来说，1–5 秒是完全可以接受的。如果你需要更低的延迟，可以缩短间隔，或者考虑使用 PostgreSQL 的 LISTEN/NOTIFY 功能，在插入新行时立即唤醒 relay。

为什么消息可能会被投递多次

你可能会想：Outbox 模式不是应该保证恰好一次投递吗？

它不能。它保证的是至少一次投递。下面是边界情况：

1. relay 成功将消息发布到 Pub/Sub。
2. 在它能够将 outbox 行更新为 'processed' 之前，relay 进程崩溃了。
3. 重启后，relay 发现消息仍然是 'pending'，于是再次发布。

这是一个罕见但可能发生的场景。处理它的标准方式是设计你的消费者具有幂等性。这意味着它们可以安全地多次接收和处理同一条消息，而不会导致错误行为。

常见的幂等策略包括：

• 使用消息的 id 作为去重键，并在操作前检查是否已经处理过。
• 使你的操作天然幂等。例如，使用 INSERT ... ON CONFLICT DO NOTHING 而不是普通的 INSERT。

替代方案：PostgreSQL 逻辑复制

上述轮询方法简单且有效，但它有两个缺点：它会引入一些延迟（最多一个轮询间隔），并且即使没有要处理的内容也会发出数据库查询。对于