Documentation
¶
Overview ¶
Package sqlbroker 是 PostgreSQL / MySQL 两个服务器型后端的共享核心:基于标准库 database/sql,把两库"真正不同"的点收进 Dialect(见 dialect.go),其余标准 SQL 一份。 实现蓝本是 sqlitebroker,但有两处 sqlite 没有、SQL 后端必须新写的机制:
- withTx 死锁重试环:sqlite 单连接串行天生不死锁,PG/MySQL 是真并发行锁, 多行事务(尤其连锁传播按树形状加锁)必然有死锁窗口,靠 Dialect.Retryable 判定后自动重跑。
- 占位符只用不复用:一律位置 ?,同值在 args 里重复传(MySQL 不支持 ?N 复用)。
认领互斥与 ReapExpired 靠 FOR UPDATE SKIP LOCKED(要求 MySQL 8.0+ / PG 9.5+), 终态更新 + 子任务唤醒同事务(宪法 III),语义由 brokertest 18 条契约统一验收(env 门控)。
本包是 internal:只给同模块的 pgbroker/mysqlbroker 薄壳 import,不对外公开。
Index ¶
- func SetTestQuotaNow(fn func() int64)
- func TotalRetries() int64
- type Broker
- func (b *Broker) Ack(ctx context.Context, id, leaseToken string, result []byte) error
- func (b *Broker) Cancel(ctx context.Context, id string) error
- func (b *Broker) Close() error
- func (b *Broker) Counts(ctx context.Context) (map[string]map[taskgate.Status]int64, error)
- func (b *Broker) Dequeue(ctx context.Context, queues []string) (*taskgate.Task, error)
- func (b *Broker) Enqueue(ctx context.Context, t *taskgate.Task) error
- func (b *Broker) Fail(ctx context.Context, id, leaseToken, errMsg string, kind taskgate.FailKind, ...) error
- func (b *Broker) FinishCanceled(ctx context.Context, id, leaseToken string) error
- func (b *Broker) Get(ctx context.Context, id string) (*taskgate.Task, error)
- func (b *Broker) Heartbeat(ctx context.Context, id, leaseToken string) error
- func (b *Broker) Init(opts taskgate.BrokerOptions) error
- func (b *Broker) List(ctx context.Context, f taskgate.Filter) ([]*taskgate.Task, error)
- func (b *Broker) QueueLen(ctx context.Context, queue string) (int, error)
- func (b *Broker) QueueQuota(queue string, qc taskgate.QueueConfig) (taskgate.QuotaGate, error)
- func (b *Broker) ReapExpired(ctx context.Context) (int, error)
- func (b *Broker) Replay(ctx context.Context, req taskgate.ReplayRequest) (*taskgate.Task, error)
- func (b *Broker) Requeue(ctx context.Context, id, leaseToken string) error
- type Config
- type Dialect
- type QuotaSQL
- type RetryClass
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func SetTestQuotaNow ¶
func SetTestQuotaNow(fn func() int64)
SetTestQuotaNow 仅测试用(与 TotalRetries 同类的测试观测口):设置介质时间覆盖, 传 nil 恢复"用数据库服务端钟"。非并发安全,只应在跑用例前的单线程阶段设置。
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker PG/MySQL 共享后端。实现 taskgate.Broker。
func New ¶
New 用打开好的 *sql.DB 和一个方言装配核心。db 由薄壳包 sql.Open 得到; 返回的 Broker 用之前必须先 Init(建表也在 Init 里,那时才有独占连接跑 DDL)。
func (*Broker) Cancel ¶
Cancel 取消:排队类状态(blocked/pending/retrying)直接 canceled 并同事务传播; running 只打 cancel_requested 标记(终态由 FinishCanceled 落库); 终态报 ErrAlreadyFinal,不存在报 ErrTaskNotFound。
func (*Broker) Dequeue ¶
Dequeue 阻塞认领:直到某队列出现"status∈{pending,retrying} 且 run_at≤now"的任务,或 ctx 取消。 循环结构与 sqlitebroker 一致,三个唤醒源:同进程写入的内部信号、注入 clock 的到点信号、ctx。 跨进程写入靠 PollInterval 兜底轮询发现(默认 200ms,一次网络往返,比 sqlite 的 100ms 放宽)。
func (*Broker) Enqueue ¶
Enqueue 入队。同一个事务内完成:ID 查重、父任务存在性校验、初始状态判定(DecideOnSubmit)、 tasks 与 task_deps 落库;生成的 ID 与判定结果回填到 *t。撞主键翻译成 ErrTaskExists。
func (*Broker) Fail ¶
func (b *Broker) Fail(ctx context.Context, id, leaseToken, errMsg string, kind taskgate.FailKind, retryAt time.Time) error
Fail 失败路径:按 FailKind 动对应计数,封顶或耗尽进 failed(同事务连锁传播), 否则进 retrying、RunAt=retryAt 到点重跑。
func (*Broker) FinishCanceled ¶
FinishCanceled worker 响应取消后收尾:running → canceled 落库并同事务传播。
func (*Broker) Heartbeat ¶
Heartbeat 续租:lease_until = now + TTL。发现取消标记时续租照做(先提交), 再返回 ErrTaskCanceled 提醒 scheduler 去 cancel handler 的 ctx。
func (*Broker) Init ¶
func (b *Broker) Init(opts taskgate.BrokerOptions) error
Init 装配运行参数(零值补默认),并建表。首次调用时在独占连接上加库级互斥锁跑 DDL, 兼容多进程冷启动并发建表(避免 PG 的 tuple concurrently updated 等脏错)。
func (*Broker) List ¶
List 按 Filter 过滤,零值字段不过滤;先过滤 → ORDER BY (created_at, id) 升序 → LIMIT/OFFSET 分页(排序分页合同见 broker-contract.md,Offset 越界返回空)。只读走连接池无锁。
func (*Broker) QueueQuota ¶
QueueQuota 构造该队列的配额闸(taskgate.QuotaProvider 能力实现)。
func (*Broker) ReapExpired ¶
ReapExpired 回收过期租约(lease_until < now 严格小于,压线不算过期)。逐行处理(不用 UPDATE...RETURNING): 先 SELECT id ... FOR UPDATE SKIP LOCKED 锁住待收割的行(多进程各跑 reaper 不互踩、不重复计), 再逐行按 sqlite 同一套语义改状态、连锁传播;封顶失败的 last_error 文案在 Go 里拼(免 || / CONCAT 方言差异)。
- 第零步:带 cancel_requested 的过期任务直接落 canceled(不占 LeaseLost),触发传播;
- 第一步:其余过期任务 LeaseLost+1,封顶进 failed(触发传播)否则回 pending;
- 第二步:防御修复——blocked 却发现父全终态的,按提交时同一套决策函数补齐。
返回值只算租约回收条数(第零步 + 第一步)。
type Config ¶
type Config struct {
TablePrefix string // 表名/索引名前缀,默认 "taskgate_";多应用共享服务器库时隔离
MaxOpenConns int // 连接池上限,默认 10;防一堆 worker 的阻塞轮询打爆 max_connections
PollInterval time.Duration // Dequeue 无果时的兜底轮询间隔,默认 200ms;跨进程写入靠它发现
MaxTxRetry int // withTx 死锁重试上限,默认 5
}
Config 共享核心的运行参数,由薄壳包从各自 Options 映射而来(零值补默认)。
type Dialect ¶
type Dialect interface {
// Name 方言名 "postgres" / "mysql",日志与错误前缀用。
Name() string
// Rebind 只改 SQL 文本、不碰 args 切片:PG 把第 n 个 ? 换成 $n;MySQL 原样返回。
// 核心里所有 SQL 一律只用不复用的位置 ?(同值在 args 里重复传),Rebind 才能保持最简形态。
Rebind(query string) string
// SchemaSQL 返回建表 + 建索引的 DDL(按 prefix 拼表名/索引名),按顺序执行。
// 各库类型映射不同(TEXT/VARCHAR、BLOB/BYTEA/LONGBLOB、INTEGER/BIGINT);
// MySQL 必须内置 utf8mb4_bin 排序规则(否则自定义 ID "abc"/"ABC" 被判重复、排序契约漂移)。
SchemaSQL(prefix string) []string
// IsDuplicateKey 判定错误是否为唯一约束冲突(主键或唯一索引)。
// PG SQLSTATE 23505 / MySQL errno 1062。实现必须 errors.As 到驱动错误类型再看码,禁止字符串匹配。
IsDuplicateKey(err error) bool
// DuplicateKeyConstraint 唯一冲突撞的是哪个约束/索引(spec 005:据此区分主键、
// uq_chain_head、uq_replay_of 三种冲突并翻译成不同的合同错误);不是唯一冲突或拿不到名字
// 返回 ""。PG 从 pgconn.PgError.ConstraintName 取(结构化字段);MySQL 驱动没有结构化字段,
// 只能从 1062 的 message 提取索引名——这是本项目唯一允许的错误字符串匹配,原因如上。
DuplicateKeyConstraint(err error) string
// IsIdempotentDDLErr 判定建表期 DDL 错误是否为"重复执行导致的良性错误"(列已存在/索引已存在)。
// PG 的 DDL 全部用 IF NOT EXISTS 表达,恒返回 false;MySQL 的 ALTER 没有 IF NOT EXISTS,
// 存量表升级(spec 005)靠 errno 1060(列已存在)/1061(索引名已存在)幂等跳过。
IsIdempotentDDLErr(err error) bool
// Retryable 判定事务错误的重试档位(见 RetryClass)。同样 errors.As 禁字符串匹配。
Retryable(err error) RetryClass
// QuotaSQL 周期配额(spec 006)两库"真正不同"的三条语句;表名按 prefix 拼好、
// 占位符用 ?(核心过 Rebind)。Release/清理语句两库同形,不在本方法(见 quota.go)。
QuotaSQL(prefix string) QuotaSQL
// Lock 建表期库级互斥:多进程首启并发跑 DDL,PG 会报 tuple concurrently updated 等脏错。
// 锁必须钉在传入的独占连接 conn 上(MySQL GET_LOCK 是会话级,连接池会错位)。
Lock(ctx context.Context, conn *sql.Conn, key string) error
// Unlock 释放建表锁(同一 conn)。PG advisory 会话结束也会自动释放,MySQL 必须显式 RELEASE_LOCK。
Unlock(ctx context.Context, conn *sql.Conn, key string) error
}
Dialect 收口 PostgreSQL 与 MySQL 两库"真正不同"的点,其余标准 SQL 全在共享核心一份。 实现放在各自薄壳包(pgbroker/mysqlbroker),因为要断言驱动私有错误类型 (*pgconn.PgError / *mysql.MySQLError),核心包 internal/sqlbroker 因此零驱动依赖。
认领(claim)与收割(reap)不在本接口:它们统一用 "SELECT ... FOR UPDATE SKIP LOCKED 拿行 → 逐行 UPDATE" 完成,这个语法两库完全一致, 不需要 UPDATE...RETURNING(PG 有 MySQL 无),核心一份代码即可。
type QuotaSQL ¶
QuotaSQL 周期配额的方言语句包(spec 006):
- Reserve 非空 = 单语句路径(PG):一条原子语句完成"取服务端时间 → 算窗口 → 检查余额 → 扣减 → RETURNING win";参数 (qkey, 时间覆盖, period, period, limit), 零行 = 耗尽;
- Reserve 为空 = 两步路径(MySQL 无 RETURNING):先 Now 取服务端时间(参数:时间覆盖, NULL = 服务端钟),Go 侧算窗口,再 Upsert 原子扣减(参数 qkey, win, limit), 按 affected rows 判定(1=插入 2=真更新 → 成功;0=没变 → 耗尽)。 取时与扣减分两步不破坏硬配额:原子性在 Upsert 上,时间只决定窗口归属, 边界竞态最多把预留记进上一窗(仍是合法窗口,不会多放)。
type RetryClass ¶
type RetryClass int
RetryClass 事务错误的重试判定,三态。sqlite 单连接串行不会死锁,所以蓝本里没有重试; PG/MySQL 是真并发行锁,多行事务(尤其连锁传播按树形状加锁)必然有死锁窗口, withTx 靠 Dialect.Retryable 判定后自动重跑整个事务(见 broker.go 的重试环)。
const ( // NotRetryable 不可重试:业务错误、约束冲突等,原样返回给调用方。 NotRetryable RetryClass = iota // RetryImmediate 立即可重试:数据库单方面 kill 的死锁/序列化失败(PG 40P01/40001、MySQL 1213), // 事务几乎瞬间返回,重试便宜,按指数退避重试到上限。 RetryImmediate // RetryLimited 有限重试:MySQL 1205 锁等待超时——默认要等 innodb_lock_wait_timeout 才报, // 不能和死锁同等指数重试(单次调用最坏挂死几百秒),只重试一次、短退避。 // mysqlbroker 同时会把会话级 innodb_lock_wait_timeout 调低配合。 RetryLimited )
Source Files