README
¶
ekit-timertask
定时任务微库——分片定时任务（FNV-1a hash、心跳续约）
本仓库是 ekit 拆分矩阵的成员（Round-4 微库化）。
安装
go get github.com/yi-nology/ekit-timertask@v0.1.0
包清单
- 根包
timertask— 基于 ekit-concurrency、ekit-logx
依赖：ekit-concurrency（async）、ekit-logx。
License
见 LICENSE
Documentation
¶
Overview ¶
Package timertask 提供延迟任务队列门面。
定义 TimerTask 接口，支持多种后端实现：
- RedisBackend: 基于 Redis ZSet 的分布式延迟任务（分片管理、心跳、重试）
- LocalBackend: 基于内存的本地延迟任务（单实例场景）
消费语义 ¶
RedisBackend 的消费模型为 at-most-once：消费流程先通过 Lua 脚本原子删除 ZSet 中的任务条目，再推入本地 worker 队列处理。如果进程在删除后、处理前 崩溃，该任务将永久丢失。应用层需自行保证幂等性。
两条兜底路径：
- 优雅停机（ctx 取消）时，本地队列中尚未处理的任务会回灌 Redis， 重启后继续消费（此路径为 at-least-once，处理前会重新投递）；
- 重试耗尽的任务会写入持久死信 ZSet（{timertask.<buzType>}.dead， score 为失败时间戳，默认保留最近 10000 条）；WithOnFailure 回调 只是死信之外的进程内通知手段，不构成持久化。
LocalBackend 为纯内存实现，进程退出即丢失所有未处理任务。
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type LocalBackend ¶
type LocalBackend struct {
// contains filtered or unexported fields
}
LocalBackend 是基于内存的本地延迟任务实现。 适用于单实例场景或开发测试环境。
func NewLocalBackend ¶
func NewLocalBackend(handler Handler, opts ...TimerTaskOption) *LocalBackend
NewLocalBackend 创建基于内存的本地延迟任务后端。
- handler: 任务处理函数
func (*LocalBackend) LocalPendingCount ¶
func (b *LocalBackend) LocalPendingCount() int
LocalPendingCount 返回本地队列中待处理任务数。
func (*LocalBackend) PendingCount ¶
func (b *LocalBackend) PendingCount(ctx context.Context) (int64, error)
PendingCount 返回队列中待处理任务数。
type RedisBackend ¶
type RedisBackend struct {
// contains filtered or unexported fields
}
RedisBackend 是基于 Redis ZSet 的分布式延迟任务实现。
func NewRedisBackend ¶
func NewRedisBackend(buzType string, client redis.UniversalClient, handler Handler, opts ...TimerTaskOption) *RedisBackend
NewRedisBackend 创建基于 Redis 的延迟任务后端。
- buzType: 业务类型标识（区分不同业务的任务）
- client: Redis 客户端（支持单机和集群）
- handler: 任务处理函数
func (*RedisBackend) LocalPendingCount ¶
func (b *RedisBackend) LocalPendingCount() int
LocalPendingCount 返回本地队列中待处理任务数。
func (*RedisBackend) PendingCount ¶
func (b *RedisBackend) PendingCount(ctx context.Context) (int64, error)
PendingCount 返回队列中待处理任务数（所有分片的 ZSet 大小之和）。
type ShardManager ¶
type ShardManager struct {
// contains filtered or unexported fields
}
ShardManager 分片管理器。 负责心跳注册、一致性哈希分配分片。
func NewShardManager ¶
func NewShardManager(buzType string, client redis.UniversalClient, totalShards int) *ShardManager
NewShardManager 创建分片管理器。
func (*ShardManager) GetClientID ¶
func (s *ShardManager) GetClientID() string
GetClientID 获取客户端 ID。
func (*ShardManager) GetMyShards ¶
func (s *ShardManager) GetMyShards() []int
GetMyShards 获取当前负责的分片列表。
func (*ShardManager) StartAssign ¶
func (s *ShardManager) StartAssign(ctx context.Context)
StartAssign 启动分片分配。
func (*ShardManager) StartHeartbeat ¶
func (s *ShardManager) StartHeartbeat(ctx context.Context)
StartHeartbeat 启动心跳注册。
type Task ¶
type Task struct {
ID string `json:"task_id"` // 任务唯一 ID
CreateTime int64 `json:"create_time"` // 创建时间（Unix 秒）
EndTime int64 `json:"end_time"` // 执行时间（Unix 秒）
Action string `json:"action"` // 任务类型/动作名
Param string `json:"param"` // 业务参数（JSON 字符串）
Extra string `json:"extra"` // 扩展字段
RetryCount int `json:"retry_count"` // 已重试次数
MaxRetry int `json:"max_retry"` // 最大重试次数
}
Task 是一个延迟任务。
type TimerTask ¶
type TimerTask interface {
// Send 提交一个延迟任务。
// task.EndTime（Unix 秒）决定任务何时被执行。
Send(ctx context.Context, task *Task) error
// Start 启动任务消费（阻塞）。
// ctx 取消时所有 goroutine 退出。
Start(ctx context.Context) error
// PendingCount 返回队列中待处理任务数。
PendingCount(ctx context.Context) (int64, error)
// LocalPendingCount 返回本地队列中待处理任务数。
LocalPendingCount() int
}
TimerTask 是延迟任务队列接口。 支持多种后端实现（Redis、本地内存等）。
type TimerTaskOption ¶
type TimerTaskOption func(*config)
TimerTaskOption 配置延迟任务的可选参数（函数选项模式）。
func WithBuckets ¶
func WithBuckets(n int) TimerTaskOption
WithBuckets 设置任务桶数量（仅 Redis 模式有效），默认 128。
func WithOnFailure ¶
func WithOnFailure(fn func(task *Task, err error)) TimerTaskOption
WithOnFailure 设置任务最终失败回调。 当任务重试次数用完后调用，可用于告警、埋点等；注意这是进程内的同步回调， 不是持久死信——RedisBackend 会自动把最终失败的任务写入 Redis 死信 ZSet， LocalBackend 没有持久死信。 默认仅记录日志。
Source Files