Documentation
¶
Overview ¶
Package redisbroker 是 Broker 的 Redis 后端:所有"多步读写必须原子"的操作 都收进单段 Lua 脚本执行(宪法 III:终态更新与子任务唤醒同一段脚本收敛), 语义以 memorybroker 为基准,由 brokertest 的 18 条契约统一验收。
关键约定(specs/002-m2-redis/research.md):
- 第 1 节:认领 = 单段 claim.lua 原子完成,不用 BLMOVE 两步;Dequeue 是 Go 侧轮询循环;
- 第 2 节:脚本内禁用 TIME/math.random,时间与 ulid 全部由 Go 经 ARGV 注入(fakeclock 生效);
- 第 3 节:传播在同一段脚本内用工作队列收敛整棵子树;
- 第 9 节:哨兵错误只在 Lua 明确判定时返回(TGERR: 错误码),网络错误原样透传, 绝不折叠成 ErrLeaseLost/ErrTaskNotFound。
键设计见 specs/002-m2-redis/data-model.md 第 2 节;脚本内用 ARGV[1] 传前缀自行拼键, 因此本后端面向单实例/主从 Redis,不支持 Redis Cluster(键不带 hash tag)。
limiter.go 分布式限流器:实现 taskgate.QueueLimiter,多进程共享同一份配额。 两层独立(与 localLimiter 的分层一致):
- 并发槽:zset 信号量 tg:sem:{q}(sem_acquire.lua 原子占槽,续期 goroutine 保活, 进程崩溃 → 续期停 → 槽按 TTL 过期自动回收,research 第 6 节);
- RPS 令牌:redis_rate 的 GCRA(research 第 7 节)。注意 redis_rate 的 Lua 用 Redis 服务器时间(redis.call TIME),属"RPS 走真时钟"的既有豁免(spec FR-018 例外), 与其余脚本"时间全由 Go 注入"的铁律并不冲突:限速要的本来就是物理时间。
键归属:tg:sem:{q} 与 redis_rate 的 rate:* 键都是限流器私有,不属 Broker 数据 (data-model.md 第 2 节),丢了最多是限流短暂失准,不影响任务数据。
Index ¶
- 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) QueueLimiter(queue string, qc taskgate.QueueConfig) (taskgate.QueueLimiter, 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 Options
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker Redis 后端。实现 taskgate.Broker。
func New ¶
New 按 Options 建连接并 PING 一次探活(配置错误尽早暴露)。 返回的 Broker 用之前必须先 Init(由 taskgate.New(cfg) 统一调用)。
func (*Broker) Cancel ¶
Cancel 取消:排队类状态(blocked/pending/retrying)直接 canceled 并同脚本传播; running 只打 cancel_requested 标记(终态由 FinishCanceled 落库); 终态报 ErrAlreadyFinal,不存在报 ErrTaskNotFound。Cancel 不带令牌(合同)。
func (*Broker) Close ¶
Close 关连接:标记 closed 并广播,让阻塞中的 Dequeue 尽快退出,再关客户端, 最后统一停掉所有已发限流器仍在续期的槽(不然续期 goroutine 会一直空转泄漏)。 顺序讲究:先关客户端再收尾限流器——归还路径的 ZREM 在关掉的连接上失败被容忍, Redis 里的槽按 TTL 自然过期回收,语义与"进程崩溃"一致(槽不许被 Close 抢先删掉, 别的进程要等 TTL 才能占,这正是崩溃自愈的既定口径)。
func (*Broker) Counts ¶
Counts 出现过的 Type×Status 稀疏矩阵:直接 HGETALL tg:stats(每段 Lua 流转时 顺手 HINCRBY 维护),O(矩阵大小),与逐个 Get 汇总一致(brokertest 契约 13 验证)。 流转来回抵消后可能留下 0 值字段,按"只含非零"的合同过滤掉。
func (*Broker) Dequeue ¶
Dequeue 阻塞认领:直到某队列出现"status∈{pending,retrying} 且 run_at≤now"的任务, 或 ctx 取消(返回 ctx.Err())。循环结构与 sqlitebroker 完全同构: 试认领一次(claim.lua 原子)→ 无果则挂起等待,三个唤醒源:
- 同进程写入踢的内部信号(Enqueue/Ack/Fail/... 后 wakeAll);
- 注入 clock 的到点信号:等 min(100ms, 最近的 delayed 到期时刻 - now), 延迟任务到点自动醒,100ms 同时兜底跨进程写入(fakeclock 下不推时间就纯挂起,不空转);
- ctx 取消。
func (*Broker) Enqueue ¶
Enqueue 入队。查重、父校验、初始状态判定、落库全部在 enqueue.lua 一段脚本内原子完成; ID 由 Go 生成注入(脚本内禁 math.random),生成的 ID 与判定结果只在成功后回填 *t—— 报错路径不能让调用方拿到一个根本不存在的孤儿 ID(合同 Enqueue 条款)。
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(hash 与 inflight score 一起续)。 发现取消标记时**续租已成**才返回 ErrTaskCanceled,scheduler 收到后 cancel handler ctx。
func (*Broker) Init ¶
func (b *Broker) Init(opts taskgate.BrokerOptions) error
Init 装配运行参数,零值补默认(TTL 60s / LeaseLostMax 3 / ThrottledMax 100 / 真时钟)。
func (*Broker) List ¶
List 按 Filter 过滤,零值字段不过滤;先过滤 → 内存按 (CreatedAt, ID) 升序 → 切 [Offset, Offset+Limit)(排序分页合同见 broker-contract.md,Offset 越界返回空)。 候选 ID 从索引集合取(Status 优先、其次 Type,否则七个状态集合并集), 剩余条件读回 hash 后在 Go 侧过滤。代价 O(候选集):分页不减少取回条数, 大库存用 Filter 缩小候选集(已知限制,research 第 5 节裁决,不建 zset 二级索引)。
func (*Broker) QueueLen ¶
QueueLen 队列积压:status∈{pending,retrying} 的数量(不看 RunAt 到没到点)。 排队中的任务要么在 pending list、要么在 delayed zset,恰好各占一处 (离开排队状态时由各脚本立刻摘除),所以 LLEN+ZCARD 就是精确值,O(1)。
func (*Broker) QueueLimiter ¶
func (b *Broker) QueueLimiter(queue string, qc taskgate.QueueConfig) (taskgate.QueueLimiter, error)
QueueLimiter 为队列构造跨进程共享的限流器(taskgate.LimiterProvider 能力接口)。 槽 TTL 复用该队列的 LeaseTTL(任务租约和槽租约同生命周期,好理解;research 第 6 节)。
func (*Broker) QueueQuota ¶
QueueQuota 构造该队列的配额闸(taskgate.QuotaProvider 能力实现)。
func (*Broker) ReapExpired ¶
ReapExpired 回收过期租约(lease_until < now 严格小于,压线不算过期): 带 cancel_requested 的直接落 canceled(不占 LeaseLost,传播);其余 LeaseLost+1, 封顶进 failed(固定文案,传播)否则回 pending 清令牌;顺带防御修复 "blocked 但父实际全部终态"的任务。整个过程在 reap.lua 一段脚本内完成, 返回值只算租约回收条数。
type Options ¶
type Options struct {
Addr string // Redis 地址,如 "127.0.0.1:6379"
Password string // 密码,空 = 不认证
DB int // 库号
// KeyPrefix 所有键的前缀,默认 "tg:";多应用共用一个 Redis 时用它隔离。
// 例外:RPS 限速状态键不在本前缀命名空间内——redis_rate 自己会再加 "rate:"
// 前缀,最终键形如 "rate:<KeyPrefix><queue>";按前缀批量清理/统计时别漏了它。
KeyPrefix string
}
Options 连接参数。库不读环境变量,应用自己填好传进来(宪法 I)。
Source Files