timertask

package module
v0.1.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 15, 2026 License: MIT Imports: 18 Imported by: 0

README

timertask

定时任务微库——分片定时任务（FNV-1a hash、心跳续约）

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

本仓库是 ekit 拆分矩阵的成员（Round-4 微库化）。

安装

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

能做什么

  • 双后端：LocalBackend（单机内存）/ RedisBackend（跨实例分片）
  • 分片协调：ShardManager 心跳 + FNV-1a 均匀分片，多实例不重不漏
  • 可靠性：任务级重试（次数+延迟）、失败回调、桶/worker 可调

使用示例

本地延迟任务
import "github.com/oekit/timertask"

var handler timertask.Handler = func(ctx context.Context, t *timertask.Task) error {
    return dispatch(t.Action, t.Param)
}
backend := timertask.NewLocalBackend(handler,
    timertask.WithRetry(3, 10*time.Second),
    timertask.WithOnFailure(func(t *timertask.Task, err error) { alert(t) }))
_ = backend.Start(ctx)
_ = backend.Send(ctx, &timertask.Task{ID: "t1", EndTime: time.Now().Add(time.Minute).Unix(), Action: "remind"})
跨实例分片（Redis）
backend := timertask.NewRedisBackend("order", redisCli, handler)
sm := timertask.NewShardManager("order", redisCli, 10) // 10 分片
sm.StartHeartbeat(ctx)
sm.StartAssign(ctx)
## 包清单

- 根包 `timertask` — 基于 [ekit-concurrency](https://github.com/oekit/concurrency)、[ekit-logx](https://github.com/oekit/logx)

依赖：[ekit-concurrency](https://github.com/oekit/concurrency)（async）、[ekit-logx](https://github.com/oekit/logx)。

## License

见 [LICENSE](LICENSE)

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

| 模块 | 内容 | 矩阵内依赖 |
|---|---|---|
| [pkg](https://github.com/oekit/pkg) | 微工具集 16 合 1（errors/validator/stringx/slicex/enumx/convert/encoding/jsonrepair/httpjson/mathx/pathx/pagination/ptrx/timex/uidgen/cache） | — |
| [concurrency](https://github.com/oekit/concurrency) | 并发与流控原语（async/group/singleflight/progress/syncx/ratelimit/retry） | — |
| [logx](https://github.com/oekit/logx) | 统一日志门面（多后端可替换 + Sink + caller 归属）+ 优雅关停 | — |
| [core](https://github.com/oekit/core) | 应用基建件（governance 治理 schema / lifecycle 契约 / network IP 工具 / version） | — |
| [istio](https://github.com/oekit/istio) | Istio Proxyless——Kitex xDS 动态熔断 | — |
| [metrics](https://github.com/oekit/metrics) | Prometheus 门面 + 弹性指标族 | [core](https://github.com/oekit/core) |
| [tracing](https://github.com/oekit/tracing) | OpenTelemetry 追踪 + trace_id 真相源 | [logx](https://github.com/oekit/logx) |
| [registry](https://github.com/oekit/registry) | 注册发现核心 + Nacos 实现 + 配置加载 + sdbridge | [core](https://github.com/oekit/core)、[logx](https://github.com/oekit/logx) |
| [data](https://github.com/oekit/data) | 缓存检查器（TTL/failTTL/inflight 合并）+ LocalCache | [concurrency](https://github.com/oekit/concurrency)、[logx](https://github.com/oekit/logx) |
| [gorm](https://github.com/oekit/gorm) | SQL/GORM 全家——连接池增强、框架无关 CRUD Service、仓储 | [logx](https://github.com/oekit/logx)、[pkg](https://github.com/oekit/pkg) |
| [redis](https://github.com/oekit/redis) | Redis 全家——客户端封装、缓存后端、分布式锁 | [concurrency](https://github.com/oekit/concurrency)、[data](https://github.com/oekit/data)、[logx](https://github.com/oekit/logx) |
| [httpx](https://github.com/oekit/httpx) | HTTP 客户端 + 负载均衡——服务发现/熔断/重试/指标/OTel 透传 | [concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[registry](https://github.com/oekit/registry)、[tracing](https://github.com/oekit/tracing) |
| [network](https://github.com/oekit/network) | 轻量网络客户端——WebSocket/NATS/MQ 抽象、连接与流工具、Octopus 推送 | [concurrency](https://github.com/oekit/concurrency)、[logx](https://github.com/oekit/logx)、[tracing](https://github.com/oekit/tracing) |
| [timertask](https://github.com/oekit/timertask) | 分片定时任务（FNV-1a、心跳续约） | [concurrency](https://github.com/oekit/concurrency)、[logx](https://github.com/oekit/logx) |
| [middleware](https://github.com/oekit/middleware) | Hertz 自研标准中间件栈 + contrib 桥接 + hmw | [concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[pkg](https://github.com/oekit/pkg)、[registry](https://github.com/oekit/registry)、[tracing](https://github.com/oekit/tracing) |
| [nns](https://github.com/oekit/nns) | NNS 服务治理客户端——provider 无关发现/配置/实例缓存 | [concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[logx](https://github.com/oekit/logx)、[registry](https://github.com/oekit/registry) |
| [nns-endpoint](https://github.com/oekit/nns-endpoint) | NNS 端点工厂——per-service 熔断、换实例重试、gRPC 池 | [concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[httpx](https://github.com/oekit/httpx)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[nns](https://github.com/oekit/nns)、[registry](https://github.com/oekit/registry) |
| [bootstrap](https://github.com/oekit/bootstrap) | 应用生命周期引导——bootstrap.Run 骨架 + healthz | [concurrency](https://github.com/oekit/concurrency)、[logx](https://github.com/oekit/logx) |
| [handler](https://github.com/oekit/handler) | HTTP handler 层——通用 CRUD Handler、参数绑定、版本端点 | [concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[gorm](https://github.com/oekit/gorm)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[middleware](https://github.com/oekit/middleware)、[pkg](https://github.com/oekit/pkg)、[registry](https://github.com/oekit/registry)、[tracing](https://github.com/oekit/tracing) |
| [hertz](https://github.com/oekit/hertz) | Hertz 运行时集成——日志桥接、生命周期、SSE/health/metrics、适配器、OTel | [bootstrap](https://github.com/oekit/bootstrap)、[concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[nns](https://github.com/oekit/nns)、[registry](https://github.com/oekit/registry)、[tracing](https://github.com/oekit/tracing) |
| [kitex](https://github.com/oekit/kitex) | Kitex 生态——日志桥接、治理热更、生命周期、BizError 桥接、kitexotel | [bootstrap](https://github.com/oekit/bootstrap)、[concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[nns](https://github.com/oekit/nns)、[pkg](https://github.com/oekit/pkg)、[registry](https://github.com/oekit/registry)、[tracing](https://github.com/oekit/tracing) |
| [workflow](https://github.com/oekit/workflow) | 组合式流程编排——串行/并行/异步/条件分支 + Hook AOP | [concurrency](https://github.com/oekit/concurrency)、[logx](https://github.com/oekit/logx) |
| [kitex-examples](https://github.com/oekit/kitex-examples) | Kitex echo 示例——client/server + IDL + 治理配置 | [bootstrap](https://github.com/oekit/bootstrap)、[concurrency](https://github.com/oekit/concurrency)、[core](https://github.com/oekit/core)、[kitex](https://github.com/oekit/kitex)、[logx](https://github.com/oekit/logx)、[metrics](https://github.com/oekit/metrics)、[nns](https://github.com/oekit/nns)、[registry](https://github.com/oekit/registry)、[tracing](https://github.com/oekit/tracing) |

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 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 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 没有持久死信。 默认仅记录日志。

func WithRetry 

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

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

func WithWorkers 

func WithWorkers(n int) TimerTaskOption

WithWorkers 设置 Worker 数量，默认 10。

Source Files

View all Source files

Directories

Path Synopsis
redis module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.