Golang轻量级事务性发件箱模式实现插件库outbox的使用

关键特性

• 轻量级: 仅添加一个外部依赖: google/uuid
• 数据库无关: 支持PostgreSQL、MySQL、Oracle等关系型数据库
• 消息代理无关: 适用于任何消息代理或外部系统
• 易于集成: 设计简单，易于集成到项目中
• 可观测性: 提供处理错误和丢弃消息的通道，可连接到监控和告警系统
• 快速发布: 可选的事务提交后立即异步发布消息，减少延迟，同时提供保证交付的备用方案
• 可配置重试和退避策略: 支持固定、指数或自定义退避策略
• 最大尝试次数保护: 自动丢弃超过配置的maxAttempts阈值的毒消息

使用方式

该库包含两个主要组件:

1. Writer: 在事务中原子性地存储实体和对应的消息
2. Reader: 在后台将存储的消息发布到消息代理

Writer

Writer确保实体和发件箱消息原子性地一起存储:

// 设置数据库连接
db, _ := sql.Open("pgx", "postgres://user:password@localhost:5432/outbox?sslmode=disable")

// 创建DBContext和Writer实例
dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
writer := outbox.NewWriter(dbCtx)

// 在业务逻辑中:
//
// 创建实体和发件箱消息
entity := Entity{
    ID:        uuid.New(),
    CreatedAt: time.Now().UTC(),
}

payload, _ := json.Marshal(entity)
metadata := json.RawMessage(`{"trace_id":"abc123","correlation_id":"xyz789"}`)
msg := outbox.NewMessage(payload,
    outbox.WithCreatedAt(entity.CreatedAt),
    outbox.WithMetadata(metadata))

// 在单个事务中写入消息和实体
err = writer.Write(ctx, msg,
    // 这个用户定义的回调在与存储发件箱消息相同的事务中执行查询
    func(ctx context.Context, txQueryer outbox.TxQueryer) error {
        _, err := txQueryer.ExecContext(ctx,
            "INSERT INTO entity (id, created_at) VALUES ($1, $2)",
            entity.ID, entity.CreatedAt,
        )
        return err
    })

Reader

Reader定期检查未发送的消息并将它们发布到消息代理:

// 创建消息发布器实现
type messagePublisher struct {
    // 你的消息代理客户端(如Kafka、RabbitMQ)
}
func (p *messagePublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    // 将消息发布到你的代理
    return nil
}

// 创建并启动reader
reader := outbox.NewReader(
    dbCtx,                              // 数据库上下文
    &messagePublisher{},                // 发布器实现
    outbox.WithInterval(5*time.Second), // 轮询间隔(默认: 10s)
    outbox.WithReadBatchSize(200),      // 读取批次大小(默认: 100)
    outbox.WithDeleteBatchSize(50),     // 删除批次大小(默认: 20)
    outbox.WithMaxAttempts(300),        // 300次尝试后丢弃(默认: MaxInt32)
    outbox.WithExponentialDelay(        // 尝试之间的延迟(默认: 指数;也可以使用固定或自定义)
        500*time.Millisecond,           // 初始延迟(默认: 200ms)
        30*time.Minute),                // 最大延迟(默认: 1h)
)
reader.Start()
defer reader.Stop(context.Background()) // 在应用关闭时停止

// 监控标准处理错误(发布/更新/删除/读取)
go func() {
    for err := range reader.Errors() {
        switch e := err.(type) {
        case *outbox.PublishError:
            log.Printf("Failed to publish message | ID: %s | Error: %v",
                e.Message.ID, e.Err)

        case *outbox.UpdateError:
            log.Printf("Failed to update message | ID: %s | Error: %v",
                e.Message.ID, e.Err)

        case *outbox.DeleteError:
            log.Printf("Batch message deletion failed | Count: %d | Error: %v",
                len(e.Messages), e.Err)
            for _, msg := range e.Messages {
                log.Printf("Failed to delete message | ID: %s", msg.ID)
            }

        case *outbox.ReadError:
            log.Printf("Failed to read outbox messages | Error: %v", e.Err)

        default:
            log.Printf("Unexpected error occurred | Error: %v", e)
        }
    }
}()

// 监控丢弃的消息(达到最大尝试次数阈值)
go func() {
    for msg := range reader.DiscardedMessages() {
        log.Printf("outbox message %s discarded after %d attempts",
            msg.ID, msg.TimesAttempted)
        // 示例后续步骤:
        //   • 转发到死信主题
        //   • 触发告警/指标
        //   • 持久化以供手动检查
    }
}()

数据库设置

1. 选择数据库方言

库支持多种关系型数据库。创建DBContext时配置适当的SQLDialect。支持的方言有PostgreSQL、MySQL、MariaDB、SQLite、Oracle和SQL Server。

// 使用MySQL方言创建DBContext的示例
dbCtx := outbox.NewDBContext(db, outbox.SQLDialectMySQL)

2. 创建发件箱表

发件箱表存储需要发布到消息代理的消息。根据你的数据库选择:

CREATE TABLE IF NOT EXISTS outbox (
    id UUID PRIMARY KEY,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    scheduled_at TIMESTAMP WITH TIME ZONE NOT NULL,
    metadata BYTEA,
    payload BYTEA NOT NULL,
    times_attempted INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_outbox_created_at ON outbox (created_at);
CREATE INDEX IF NOT EXISTS idx_outbox_scheduled_at ON outbox (scheduled_at);

CREATE TABLE IF NOT EXISTS outbox (
    id BINARY(16) PRIMARY KEY,
    created_at TIMESTAMP(3) NOT NULL,
    scheduled_at TIMESTAMP(3) NOT NULL,
    metadata BLOB,
    payload BLOB NOT NULL,
    times_attempted INT NOT NULL
);

CREATE INDEX idx_outbox_created_at ON outbox (created_at);
CREATE INDEX idx_outbox_scheduled_at ON outbox (scheduled_at);

示例

完整的工作示例:

• Postgres & Kafka
• Oracle & NATS
• MySQL & RabbitMQ

运行示例:

cd examples/postgres-kafka # 或examples/oracle-nats或examples/mysql-rabitmq
../../scripts/up-and-wait.sh
go run service.go

# 在另一个终端触发POST请求创建实体
curl -X POST http://localhost:8080/entity

常见问题

当我的服务的多个实例使用该库时会发生什么?

当运行多个服务实例时，每个实例都有自己的reader，请注意:

• 多个reader会独立检索消息，可能导致消息被多次发布。处理方式:
  1. 确保消费者是幂等的并接受重复
  2. 如果可用，使用代理的重复数据删除功能
  3. 仅在单个实例中运行reader

乐观发布功能可以显著减少重复数量。

使用pgxpool时如何实例化DBContext?

可以使用stdlib.OpenDBFromPool函数从*pgxpool.Pool获取*sql.DB。

import (
    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/jackc/pgx/v5/stdlib"
    "github.com/oagudo/outbox"
)

// ...
pool, _ := pgxpool.New(context.Background(), os.Getenv("DATABASE_URL"))
db := stdlib.OpenDBFromPool(pool)
dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)