如何使用 Go 语言构建一个基于 PostgreSQL 的作业队列系统

Go语言中的接口

Go语言中的接口用于描述对象的行为。对于Swig框架来说，它并不需要知道每个工作者的具体类型；它只需要确保这些工作者能够提供作业名称，并且能够执行相应的作业即可：

type Worker interface {
    JobName() string
    Process(context.Context) error
}

如果某个类型具备这些方法，那么它就无需进行显式声明即可满足该接口的要求。这就是为什么在 Go 语言中接口如此有用的原因之一——它们允许人们根据行为来设计程序，而不是依赖继承机制。

如何在不共享状态的情况下注册工作线程

Swig 提供了一个工作线程注册系统，该系统会将作业名称与相应的工作者类型关联起来：

registry := workers.NewWorkerRegistry()
registry.RegisterWorker(&EmailWorker{})

后来，当某条作业记录中包含 kind = 'send_email' 时，Swig 会查找已注册的工作者类型并启动它来执行相应的任务。

不过这里存在一个潜在的并发问题：如果注册系统存储的是指向 &EmailWorker{} 的指针，并将其重复用于多个作业，那么多个 goroutine 可能会同时尝试将 JSON 数据解码为同一个 Go 对象，从而导致数据混乱。

Swig 通过内部采用工厂模式来避免这种问题。在注册新作业时，系统会创建一个新的工作者实例，而不会重复使用现有的指针。这样一来，虽然公共 API 保持简单，但内部实现却能确保安全性。

如何添加作业

从用户的角度来看，添加作业的操作过程如下：

err := swigClient.AddJob(ctx, &EmailWorker{
    To:      "user@example.com",
    Subject: "Welcome!",
    Body:    "Thanks for signing up.",
})

在 Swig 的内部，添加作业的具体流程大致如下：

argsJSON, err := json.Marshal(workerWithArgs)
if err != nil {
    return err
}

_, err = db.ExecContext(ctx, `
    INSERT INTO swig_jobs (kind, queue, payload, priority, scheduled_for, status)
    VALUES (\(1, \)2, \(3, \)4, $5, 'pending')
`, jobName, queue, argsJSON, priority, runAt)

如何在事务中添加作业

使用 PostgreSQL 来处理作业的一个最大优势就是它支持事务机制，从而可以确保操作的原子性。

想象一下，当有用户注册时，你需要同时完成两个操作：将用户信息插入数据库，并发送欢迎邮件。如果这两个操作分开执行，就可能会出现状态不一致的情况。而使用事务的话，这两个操作要么都成功，要么都失败：

tx, err := pool.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(ctx)

_, err = tx.Exec(ctx, `INSERT INTO users (email) VALUES ($1)`, email)
if err != nil {
    return err
}

err = swigClient.AddJobWithTx(ctx, tx, &EmailWorker{
    To:      email,
    Subject: "Welcome!",
    Body:    "Thanks for joining.",
})
if err != nil {
    return err
}

return tx.commit(ctx)

如果事务被回滚，那么用户信息就不会被插入数据库，邮件也不会被发送。而当数据库和消息队列是分开的系统时，要保证这种一致性就变得非常困难了。

如何安全地管理多个工作线程

当多个工作者同时运行时，队列机制就会变得有趣起来。想象一下，有三个工作者都向 PostgreSQL 请求下一项待处理的任务——你肯定不希望这三个工作者处理同一项任务。

如果采用一种简单直接的方法，就会出现竞争条件：两个工作者可能会在选择到同一项任务之后，才去更新它。

PostgreSQL 的 FOR UPDATE SKIP LOCKED 机制

PostgreSQL 可以锁定在事务中选中的行。`FOR UPDATE` 表示“因为我打算更新这一行，所以要锁定它”；而 `SKIP LOCKED` 则表示“如果已经有其他工作者锁定了这一行，那就跳过它，去选择另一行”。

这种机制非常适合处理队列任务：

• 工作者 A 锁定了任务 1

• 工作者 B 跳过了任务 1，锁定了任务 2

• 工作者 C 跳过了任务 1 和任务 2，锁定了任务 3

在这种情况下，根本不需要中央协调器。Swig 使用的是原子性更新机制：

UPDATE swig_jobs
SET status = 'processing',
    instance_id = $1,
    worker_id = $2,
    locked_at = NOW(),
    attempts = attempts + 1
WHERE id = (
  SELECT id
  FROM swig_jobs
  WHERE status = 'pending'
    AND scheduled_for <= NOW()
  ORDER BY priority DESC, created_at
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
RETURNING id, kind, payload;

这条查询会找到一项待处理的任务，跳过那些已经被锁定的任务，将其标记为“正在处理中”，记录下是哪个工作者选择了这项任务，然后返回该任务的详细信息。所有这些操作都是原子性完成的——工作者们不需要先执行 `SELECT` 操作，然后再等待后面的 `UPDATE` 操作，因为这一切都会同时完成。

如何使用协程来实现并发处理

Swig 会以协程的形式启动各个工作者的运行线程：

for i := 0; i < maxWorkers; i++ {
    go s.startWorker(ctx, queueType)
}

每个工作者都是独立运行的。PostgreSQL 负责分配任务给各个工作者；而 Go 语言则通过协程来实现并发处理，同时 PostgreSQL 也利用锁定机制来确保任务分配的公平性。

如何实现优雅的系统关闭

当一个服务需要关闭时，它应该等待所有工作者完成当前的工作后再进行关闭操作。Go 语言中的 `sync.WaitGroup` 可以帮助实现这一点：

var wg sync.WaitGroup

wg.Add(1)
go func() {
    defer wg.Done()
    processJobs()
}()

wg.Wait()

Swig 还使用了 `sync.Once` 来确保关闭操作是幂等的——即使多次调用 `Stop` 方法，也不会因为通道被重复关闭而导致程序异常。在生产环境中，系统的关闭流程往往与测试环境中的情况有所不同。

如何使用 LISTEN/NOTIFY 机制来唤醒工作者

如果工作者们不断地从数据库中查询任务信息，那么当队列为空时，它们就会浪费资源。PostgreSQL 提供了 `LISTEN/NOTIFY` 机制来解决这个问题。

一个连接可以通过监听某个通道来接收新的任务请求：

LISTEN swig_jobs;

另一个会话可以发送通知：

NOTIFY swig_jobs, '{"id":"job-id"}';

Swig会创建一个触发器，这样当有作业被插入数据库时，PostgreSQL就会发送通知。在没有任务需要处理时，工作进程会处于等待状态；而当有新的作业到来时，这些进程便会开始执行。

这里有一个关于PostgreSQL的重要细节：LISTEN命令是会话级别的。因此，工作进程必须在与执行LISTEN命令的同一数据库会话中等待通知。Swig通过为每个工作进程创建一个专用的监听器来处理这个问题；这个监听器会在该工作进程的整个生命周期内一直保持活跃状态。

这是一个常见的后端开发经验：像连接池这样的抽象机制确实非常有用，但某些数据库功能实际上依赖于特定连接的生命周期。

如何使用建议锁来选举领导者

有些队列维护任务应该一次只在一个实例上运行，这些任务包括重试失败的任务、恢复已暂停的任务以及清理旧的历史记录。

Swig就是利用PostgreSQL的建议锁来实现这一点的：

SELECT pg_try_advisory_lock($1);

如果返回的结果为true，那么该Swig实例就会成为领导者。建议锁也是会话级别的，因此Swig会为领导者的角色专门创建一个连接来使用建议锁。如果这个会话结束，PostgreSQL会释放这个锁，然后另一个实例就可以接替它的领导职责。这样的机制实现了简单的故障转移功能，而且不需要借助ZooKeeper或etcd这样的工具。

如何处理失败的任务

当某个工作进程返回错误信息时，Swig会记录下这个错误，并要么重新尝试执行该任务，要么将其标记为失败：

UPDATE swig_jobs
SET status = CASE
    WHEN attempts >= max_attempts THEN 'failed'
    ELSE 'pending'
  END,
  last_error = $2,
  last_error_at = NOW()
WHERE id = $1;

关于交付语义的说明

人们很容易认为作业队列会恰好处理一次每个任务。但在分布式系统中，这种说法其实是危险的。

请考虑以下这种情况：

1. 一个工作进程发送了一封电子邮件

2. 但这个工作进程在完成标记之前就崩溃了

3. 系统会重新尝试执行这项任务

4. 那么这封电子邮件可能会被再次发送

更准确的描述是：Swig提供了原子的任务获取机制以及至少一次的处理保证。由于任务可以被重新尝试执行，因此工作进程必须是幂等的——多次执行相同的操作应该会得到相同的结果。

如何对数据库驱动程序进行抽象

Swig通过一个驱动接口同时支持pgx和database/sql》这两种数据库驱动程序：

type Driver interface {
    Exec(ctx context.Context, sql string, args ...interface{}) error
    Query(ctx context.Context, sql string, args ...interface{}) (Rows, error)
    QueryRow(ctx context.Context, sql string, args ...interface{}) Row
    WithTx(ctx context.Context, fn func(tx Transaction) error) error
    NewListener(ctx context.Context, channel string) (Listener, error)
    TryAdvisoryLock(ctx context.Context, lockID int64) (AdvisoryLock, bool, error)
}

核心队列代码仅依赖于具体的功能实现，而与特定的库无关。这种设计在Go语言中非常常见：首先明确核心组件所需的功能，然后为具体的依赖关系编写相应的适配代码，从而确保核心逻辑的独立性。

结论

对于并非所有系统而言，以PostgreSQL作为后端的队列解决方案并不一定是最佳选择。如果需要处理大量事件流，Kafka可能更为适合；而如果需要实现复杂的路由机制，RabbitMQ则会更加适用。
然而，对于许多使用Go语言开发的应用程序来说，PostgreSQL早已成为了一个成熟且实用的选择。Swig项目证明了：只需借助简单的Go API以及PostgreSQL的一些功能，就能构建出高效的后台处理系统。

• 将任务信息存储在数据库表中

• 使用`FOR UPDATE SKIP LOCKED`语句原子性地获取任务

• 通过专用的`LISTEN/NOTIFY`会话来唤醒工作线程

• 利用建议锁机制来协调各个组件的运行顺序

通过事务确保应用程序的数据和任务状态的一致性

• 利用goroutine和context机制来管理工作线程的生命周期

这种组合为后台处理任务提供了坚实的基础，同时也是了解Go语言与PostgreSQL如何在生产环境中协同工作的绝佳范例。您可以通过github.com/glamboyosa/swig查看该项目的完整源代码。

Related Posts:

• PostgresML：通过 PostgreSQL 集成简化 AI 模型部署
• 如何在 Go 语言和 PostgreSQL 中实现“待发邮件模式”
• 如何将 PostgreSQL 用作缓存、队列及搜索引擎
• 如何使用 ScyllaDB NoSQL 和 NextJS 构建低延迟视频流应用程序
• 简化数据管理：StorageX 如何使用 AI 帮助开发人员
• 如何使用 PhoneGap SDK 安装实时聊天和聊天机器人