timertask

README

timertask

分片定时任务——秒级延时/跨实例 Redis 后端

🌐 项目主页：https://oekit.github.io/ · 名称由来：Open Easy Kit（开源 · 易用 · 工具集合） 🤖 AI Agent：本仓规范与硬规则见 AGENTS.md；机器可读索引见 llms.txt（接口签名以源码为准）。

安装

go get github.com/oekit/timertask@v0.1.9

这个仓库做什么

timertask 是分片定时任务：按任务 ID 哈希分片、支持秒级精度的延时任务、 心跳续约防重复执行。提供 Local（进程内）与 Redis（跨实例）两种后端。

快速使用

import "github.com/oekit/timertask"

var handler timertask.Handler = func(ctx context.Context, t *timertask.Task) error {
    return dispatch(t.Action, t.Param) // Action/Param 由业务自定义
}

backend := timertask.NewLocalBackend(handler)
_ = backend.Start(ctx)
_ = backend.Send(ctx, &timertask.Task{
    ID: "t1", EndTime: time.Now().Add(time.Minute).Unix(),
    Action: "remind", Param: `{"uid":42}`,
})

边界

• 跨实例去重需要 Redis → NewRedisBackend；cron 表达式解析在 pkg/timex/cron

矩阵总览（24 活跃模块，自动生成）

模块	内容	矩阵内依赖
pkg	微工具集 19 合 1（errors/bloomx/contextx/container/stringx/slicex/enumx/convert/encoding/jsonrepair/httpjson/mathx/pathx/pagination/ptrx/timex/cache/bufiox/strmap；v0.10.0 起 uidgen/validator 拆至 kitx）；cache（v0.4.0 起）含缓存接口 CacheAside + BatchGetter 批量能力（MGet）+ LocalCache/SingleFlightCache + cachetest 契约 + Checker；container（v0.9.0 起，收编原 collectionx）泛型容器 Set/Ring/Queue/SafeMap + Map（sync.Map 背书）/Slice（并发安全）；timex/stringx（v0.7.0 起）时间段边界 BeginningOf/EndOf 七对（分/时/日/周/月/季/年，周起始显式参数）+ trie 多模式替换器 Replacer（最长匹配）/ NewMask 敏感词遮罩；singleflighttest（v0.8.0 起）singleflight 共享契约测试套件（internal 与 concurrency 双侧引用）	—
concurrency	并发与流控原语（async/gopool/group/singleflight/progress/syncx/ratelimit/retry）+ window 滑动窗口 / breaker 自适应熔断（规范哨兵 ErrCircuitOpen）/ shedding CPU 自适应降载 / batch 攒批执行器（满条数/满字节/窗口去重）/ timewheel 单级时间轮；v0.5.0 起 retry 指数退避+抖动（MaxDelay 封顶/溢出饱和）、loop 周期原语（忙时丢 tick/单轮 panic 收编/FirstDelay 可打断）、async labeled RunSafe（panic 按任务标签上报）、syncx.FanInErr 逐槽错误（panic 槽显式标记）、KeyedLimiter 增强（容量护栏/RetryAfter/速率热更）并接入 pkg/singleflighttest 契约测试	pkg
logx	统一日志门面（多后端可替换 + Sink + caller 归属）+ 优雅关停；桥接子包 hlog/hertzlog/kitexlog 为嵌套模块按需引入	—
core	应用基建件（governance 治理 schema / config 配置加载 + default/env/range/options tag 校验 / lifecycle 契约 / network IP 工具 / version）	—
kitx	标识与校验——uidgen（Snowflake 全局单例+Extract 元数据、手写 RFC 4122 v4 UUID 零第三方依赖、RandomHex/RandomString 拒绝采样随机串）+ validator（go-playground/validator v10 接口化门面：struct tag 校验/自定义规则注册/ValidationError Unwrap 为 BizError 400 对接 middleware.Fail），自 pkg v0.10.0 拆出	pkg
metrics	Prometheus 门面 + 弹性指标族	core
tracing	OpenTelemetry 追踪 + trace_id 真相源	logx
registry	注册发现核心 + Nacos 实现 + Nacos 配置源 + sdbridge	core、logx
data	数据访问门面——复刻 ekit data/ 能力面：cache/rediscache/crud/distlock/gormx/redisx/timertask/cachetest 八子包 re-export（类型同一，实现住 pkg 与各子仓）+ cached 缓存读写编排（写后失效/事务提交后失效/查询缓存版本号失效/GetByIDs 批量点读 MGet 快路径/bloom 防穿透）+ twolevel 两级缓存与跨实例失效广播（v0.7.0 起本仓唯二非 re-export 实现）；all/ 聚合模块已废弃 retract	concurrency、gormx、logx、pkg、redisx、timertask
gormx	SQL/GORM 全家——连接池增强、框架无关 CRUD Service、仓储（原 oekit/gorm，v0.3.0 更名）	logx、pkg
redisx	Redis 全家——客户端封装（根包）、分布式缓存 cache/、分布式锁 distlock、分布式限流 ratelimit（Lua 三态 + 故障本地兜底）	concurrency、logx、pkg
etcdx	etcd 全家——客户端封装（根包：生产默认值/Health 探活/Close(ctx) 幂等）、registry 注册发现实现（SPI 对接 registry/nacos 链路：Lease KeepAlive 注册 + 丢约指数退避自动重注册、prefix Watch 订阅推送、ConfigStore KV 配置监听）、distlock 分布式锁（Session+concurrency.Mutex 公平排队，接口与 redisx/distlock.Locker 同形可互换，可选无 etcd 降级放行）	concurrency、core、logx、registry
httpx	HTTP 客户端 + 负载均衡——服务发现/熔断（规范哨兵 ErrCircuitOpen）/重试/指标/OTel 透传/一致性哈希（权重感知虚拟副本）	concurrency、core、logx、metrics、registry、tracing
network	轻量网络客户端——WebSocket/NATS/MQ 抽象、连接与流工具	concurrency、logx、tracing
timertask	分片定时任务（FNV-1a、心跳续约）；v0.1.5 起可选可靠投递 DeliveryAtLeastOnce（租约 + Ack + 可见性超时；过期租约回扫重投/死信、停机 best-effort 收租，v0.1.7 接入生命周期真正生效）	concurrency、logx
middleware	Hertz 自研标准中间件栈 + contrib 桥接 + hmw + SheddingMiddleware 自适应降载；v0.6.0 起 accesslog 凭证路径脱敏（WithAccessLogRedactPathPrefixes）、ratelimit 默认 KeyFunc 去端口（连接桶→地址桶）	concurrency、core、logx、metrics、pkg、registry、tracing
k8s	Kubernetes 客户端封装——RESTConfig/Clientset/DynamicClient 构建（显式 kubeconfig → in-cluster 自动 → KUBECONFIG/~/.kube/config 逐级回退，QPS/Burst 限流默认 50/100）+ leaderelection 子包 Lease 选主简化封装（OnStartedLeading/OnStoppedLeading/OnNewLeader 回调，主备单活，仅要求 Lease CRUD 权限）	logx
nns	NNS 服务治理客户端——provider 无关发现/配置/实例缓存	concurrency、core、logx、registry
nns-endpoint	NNS 端点工厂——per-service 熔断、换实例重试、gRPC 池	concurrency、core、httpx、logx、metrics、nns、registry
bootstrap	应用生命周期引导——bootstrap.Run 骨架 + healthz	concurrency、logx
hertz	Hertz 运行时集成——日志桥接、生命周期、SSE/health/metrics、适配器、OTel、crudh/validatorh（基于 kitx/validator）/versionh（原 handler 微库收编）	bootstrap、concurrency、core、gormx、kitx、logx、metrics、middleware、nns、pkg、registry、tracing
kitex	Kitex 生态——日志桥接、治理热更、生命周期、BizError 桥接、kitexotel、istio（xDS 动态熔断，原 istio 微库收编）	bootstrap、concurrency、core、logx、metrics、nns、pkg、registry
workflow	组合式流程编排——串行/并行/异步/条件分支 + Hook AOP	concurrency、logx
kitex-examples	Kitex echo 示例——client/server + IDL + 治理配置	bootstrap、concurrency、core、kitex、logx、metrics、nns、pkg、registry、tracing

依赖严格自底向上；微库已合并为 pkg，旧微库仓已删除。

本仓在矩阵中的位置与依赖见上表「矩阵内依赖」列。历史沿革：源自 ekit 单体库 （git.enjoye.top/enjoydream/ekit v0.34.0，冻结维护），2026-10 拆分为独立 module。

License

MIT — 见 LICENSE

Documentation

Overview

Package timertask 提供延迟任务队列门面。

定义 TimerTask 接口，支持多种后端实现：

• RedisBackend: 基于 Redis ZSet 的分布式延迟任务（分片管理、心跳、重试）
• LocalBackend: 基于内存的本地延迟任务（单实例场景）

消费语义

RedisBackend 支持两种投递模式（WithDeliveryMode / WithAtLeastOnce）：

• DeliveryAtMostOnce（默认，与历史行为一致）：消费流程先通过 Lua 脚本 原子删除 ZSet 中的任务条目，再推入本地 worker 队列处理。进程在删除后、 处理前崩溃，任务永久丢失。应用层需自行保证幂等性。
• DeliveryAtLeastOnce（可靠模式）：claim 时把条目移入 processing 租约集 （score=租约到期），handler 成功后 Ack 删除；租约超时由回扫按 RetryCount 重投。kill -9 后未 Ack 任务会在 visibility timeout 后被重投（可容忍重复， 由调用方幂等）。

兜底路径：

• 优雅停机（ctx 取消）时，本地队列中尚未处理的任务会回灌 Redis， 重启后继续消费；
• 重试耗尽的任务会写入持久死信 ZSet（{timertask.<buzType>}.dead， score 为失败时间戳，默认保留最近 10000 条）；WithOnFailure 回调 只是死信之外的进程内通知手段，不构成持久化。

哪种任务选哪种模式：

• 提醒/审计/计费等不可丢事件 → DeliveryAtLeastOnce + 调用方幂等；
• 可丢弃的指标/通知/缓存预热 → DeliveryAtMostOnce（更省 Redis 往返）。

LocalBackend 为纯内存实现，进程退出即丢失所有未处理任务，不适合承载不可丢任务。

Index

• type DeliveryMode
• type Handler
• type LocalBackend
  • func NewLocalBackend(handler Handler, opts ...TimerTaskOption) *LocalBackend
  • func (b *LocalBackend) LocalPendingCount() int
  • func (b *LocalBackend) PendingCount(ctx context.Context) (int64, error)
  • func (b *LocalBackend) Send(ctx context.Context, task *Task) error
  • func (b *LocalBackend) Start(ctx context.Context) error
• type RedisBackend
  • func NewRedisBackend(buzType string, client redis.UniversalClient, handler Handler, ...) *RedisBackend
  • func (b *RedisBackend) LocalPendingCount() int
  • func (b *RedisBackend) PendingCount(ctx context.Context) (int64, error)
  • func (b *RedisBackend) Send(ctx context.Context, task *Task) error
  • func (b *RedisBackend) Start(ctx context.Context) error
• type ShardManager
  • func NewShardManager(buzType string, client redis.UniversalClient, totalShards int) *ShardManager
  • func (s *ShardManager) GetClientID() string
  • func (s *ShardManager) GetMyShards() []int
  • func (s *ShardManager) StartAssign(ctx context.Context)
  • func (s *ShardManager) StartHeartbeat(ctx context.Context)
• type Task
• type TimerTask
• type TimerTaskOption
  • func WithAtLeastOnce(visibilityTimeout time.Duration) TimerTaskOption
  • func WithBuckets(n int) TimerTaskOption
  • func WithDeliveryMode(m DeliveryMode) TimerTaskOption
  • func WithOnFailure(fn func(task *Task, err error)) TimerTaskOption
  • func WithRetry(count int, delay time.Duration) TimerTaskOption
  • func WithVisibilityTimeout(d time.Duration) TimerTaskOption
  • func WithWorkers(n int) TimerTaskOption

Types

type DeliveryMode

type DeliveryMode int

DeliveryMode 是任务投递语义。

const (
	// DeliveryAtMostOnce 默认：claim 时从 ZSet 原子删除后处理。
	// 删除后、处理前崩溃 = 任务永久丢失（仅优雅停机回灌兜底）。
	DeliveryAtMostOnce DeliveryMode = iota
	// DeliveryAtLeastOnce 可靠模式：claim 时移入 processing 租约集，
	// handler 成功后 Ack 删除；租约超时由回扫重投。可容忍重复，调用方幂等。
	DeliveryAtLeastOnce
)

type Handler

type Handler func(ctx context.Context, task *Task) error

Handler 是任务处理函数。

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 返回队列中待处理任务数。

func (*LocalBackend) Send

func (b *LocalBackend) Send(ctx context.Context, task *Task) error

Send 提交一个延迟任务。 task.EndTime（Unix 秒）决定任务何时被执行。

func (*LocalBackend) Start

func (b *LocalBackend) Start(ctx context.Context) error

Start 启动任务消费（阻塞）。 ctx 取消时所有 goroutine 退出。

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 大小之和）。

func (*RedisBackend) Send

func (b *RedisBackend) Send(ctx context.Context, task *Task) error

Send 提交一个延迟任务。 task.EndTime（Unix 秒）决定任务何时被执行。

func (*RedisBackend) Start

func (b *RedisBackend) Start(ctx context.Context) error

Start 启动任务消费（阻塞）。 ctx 取消时所有 goroutine 退出。

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 WithAtLeastOnce

func WithAtLeastOnce(visibilityTimeout time.Duration) TimerTaskOption

WithAtLeastOnce 开启可靠投递（等价 WithDeliveryMode(DeliveryAtLeastOnce)）， visibilityTimeout 为 processing 租约时长（<=0 取默认 60s）。 claim 后任务进入租约集，成功 Ack 删除；超时未 Ack 由回扫按 RetryCount 重投。

func WithBuckets

func WithBuckets(n int) TimerTaskOption

WithBuckets 设置任务桶数量（仅 Redis 模式有效），默认 128。

func WithDeliveryMode

func WithDeliveryMode(m DeliveryMode) TimerTaskOption

WithDeliveryMode 设置投递语义。默认 DeliveryAtMostOnce（与历史行为一致）。 DeliveryAtLeastOnce 时可用 WithVisibilityTimeout 调整租约时长。 仅 RedisBackend 生效；LocalBackend 为纯内存，无持久投递语义。

func WithOnFailure

func WithOnFailure(fn func(task *Task, err error)) TimerTaskOption

WithOnFailure 设置任务最终失败回调。 当任务重试次数用完后调用，可用于告警、埋点等；注意这是进程内的同步回调， 不是持久死信——RedisBackend 会自动把最终失败的任务写入 Redis 死信 ZSet， LocalBackend 没有持久死信。 默认仅记录日志。

func WithRetry

func WithRetry(count int, delay time.Duration) TimerTaskOption

WithRetry 设置任务重试次数和延迟，默认 3 次，5 秒延迟。

func WithVisibilityTimeout

func WithVisibilityTimeout(d time.Duration) TimerTaskOption

WithVisibilityTimeout 设置 at-least-once 模式的租约时长（<=0 取默认 60s）。

func WithWorkers

func WithWorkers(n int) TimerTaskOption

WithWorkers 设置 Worker 数量，默认 10。