Documentation
¶
Overview ¶
worker 包选主原语：多副本部署时"只能跑一份"的组件（出站轮询、定时清理、 通知汇总等）经租约竞争，与 DB 即队列的 worker 池组成完整的横向扩展故事—— 任务面靠 ClaimNextPending 原子抢占天然分片，控制面靠租约选主收敛为单份。
选主语义：任期 = ttl；持有者失联（进程崩溃）后 ≤ttl 自动换主； 优雅停机主动让位（Release）；脑裂窗口 ≤ 一个轮询间隔，调用方组件需容忍 短暂双主（Argus 轮询任务源幂等双查，双跑无害只浪费）。
Package worker DB 即队列 worker pool。 webhook/API 同步路径只落库（pending）即返回，N 个 worker 从表拉 pending（原子抢占）执行。 心跳续期 + 过期重置 + 两阶段优雅停机。
Index ¶
Constants ¶
const ( HeartbeatInterval = 20 * time.Second StaleRunningAfter = 90 * time.Second )
默认参数。
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type LeaderElector ¶
type LeaderElector struct {
// contains filtered or unexported fields
}
LeaderElector 租约选主器：周期 TryAcquire 竞争，IsLeader 反映最近一次结果， 角色翻转时触发 onGained/onLost 回调。
func NewLeaderElector ¶
func NewLeaderElector(store LeaseStore, key, holder string, ttl, interval time.Duration, opts ...LeaderOption) *LeaderElector
NewLeaderElector 创建选主器。interval 建议 ttl/3（失主换主延迟 ≈ ttl + interval）。
func (*LeaderElector) IsLeader ¶
func (e *LeaderElector) IsLeader() bool
IsLeader 当前是否持有领导权（最近一次续约结果）。
func (*LeaderElector) Start ¶
func (e *LeaderElector) Start(ctx context.Context)
Start 启动竞选循环：立即竞争一次（缩短冷启动等待），此后按 interval 续约。 ctx 取消或 Stop() 退出；退出前若持有租约则主动让位（计划内交接，不触发 onLost）。 重复调用为 no-op（只允许一个竞选 goroutine）。
func (*LeaderElector) Stop ¶
func (e *LeaderElector) Stop()
Stop 停止竞选并主动让位（幂等；优雅停机的计划内交接，不触发 onLost）。 等待竞选 goroutine 退出后返回——返回后 IsLeader 必为 false 且租约已让出。 Start 未被调用时立即返回（无租约可让）。
type LeaderOption ¶
type LeaderOption func(*LeaderElector)
LeaderOption 选主器可选项。
func WithLeaderLogger ¶
func WithLeaderLogger(log logx.Logger) LeaderOption
WithLeaderLogger 日志（nil 安全，缺省静默）。
func WithOnLost ¶
func WithOnLost(f func()) LeaderOption
WithOnLost 失去领导权回调（续约失败；Stop 主动让位不触发——那是计划内交接）。
type LeaseStore ¶
type LeaseStore interface {
// TryAcquire 原子获取或续约租约。key 空闲/已过期/持有者就是 holder →
// 置 holder 并顺延 ttl，返回 true；他人持有未过期 → 返回 false。
TryAcquire(ctx context.Context, key, holder string, ttl time.Duration) (bool, error)
// Release 释放租约（仅持有者本人生效；优雅停机让位）。
// 非持有者或 key 不存在 → no-op 返回 nil。
Release(ctx context.Context, key, holder string) error
}
LeaseStore 租约存储接口（调用方按存储实现：SQL 一行条件 UPSERT 即可）。 语义必须原子：TryAcquire 的"读-判-写"不允许并发穿插（SQL 用 `UPDATE ... WHERE key=? AND (holder=? OR expires_at<?)` 条件更新实现）。
type Pool ¶
type Pool struct {
Queue TaskQueue
// Run 单任务执行体。
Run func(ctx context.Context, taskID string) error
N int
Log logx.Logger
// ClaimTimeout 单次 ClaimNextPending 的挂死兜底（0 = 5 分钟缺省）。
// Queue 是调用方实现（GORM 等），与 runTask/claim 的 panic 隔离同一不变量：
// 外部代码异常不得让 worker 永久僵死——超时按 claim 失败退避重试。
ClaimTimeout time.Duration
// contains filtered or unexported fields
}
Pool worker 池。
type TaskQueue ¶
type TaskQueue interface {
// ClaimNextPending 原子抢占下一个 pending 任务（pending → running）。
ClaimNextPending(ctx context.Context) (taskID string, ok bool, err error)
// TouchRunningHeartbeats 续期本实例 running 任务的心跳。
TouchRunningHeartbeats(ctx context.Context) error
// ResetRunningToPending 重置心跳已死的 running 任务回 pending。
ResetRunningToPending(ctx context.Context, staleAfter time.Duration) (int64, error)
}
TaskQueue 任务队列存储接口（调用方实现，如 GORM/MySQL/Redis）。
Source Files