README
¶
task: High-Performance In-Memory Task Runner
task is a zero-dependency, concurrency-controlled in-memory task execution library for Go.
It bridges the gap between raw go func() (which is uncontrolled and risky) and heavy distributed queues (like RabbitMQ/Redis). It is designed to provide Bulkheading (resource isolation) and Fault Tolerance (panic recovery) for your Go services.
Why use task?
- Protection: Limits the number of concurrent goroutines to prevent OOM (Out of Memory).
- Fault Tolerance: Automatically catches panics in tasks to keep the main process alive.
- Graceful Shutdown: Ensures all running tasks are completed before the server exits (Drain pattern).
- Observability: Provides real-time metrics (Active, Queued, Processed, Panics).
- Zero Dependency: Built using only the Go standard library.
Installation
go get github.com/oy3o/task
Quick Start
1. Simple Safe Goroutine (No Pool)
If you just want to run a background job safely without managing a pool:
import "github.com/oy3o/task"
func main() {
// Replaces 'go func()'
// Automatically recovers from panics and logs stack traces
task.Go(context.Background(), func(ctx context.Context) {
doWork()
}, nil)
}
2. Worker Pool (Concurrency Limit)
For high-throughput scenarios, use a Runner to limit concurrency.
func main() {
// Create a runner with 10 workers and a buffer queue of 1000
r := task.NewRunner(
task.WithMaxWorkers(10),
task.WithQueueSize(1000),
task.WithErrorHandler(func(ctx context.Context, p any) {
fmt.Printf("Task panicked: %v\n", p)
}),
)
// Start workers
r.Start(context.Background())
// Submit tasks (Non-blocking)
err := r.Submit(func(ctx context.Context) {
// Do heavy work here...
time.Sleep(time.Second)
})
if err == task.ErrQueueFull {
// Handle backpressure (e.g., return 503 to client)
}
// Graceful Stop (Waits for running tasks to finish)
r.Stop(context.Background())
}
3. Synchronous Wait (SubmitAndWait)
Useful when you need concurrency limits but need the result immediately (e.g., batch processing in an HTTP request).
err := r.SubmitAndWait(ctx, func(ctx context.Context) {
// process item...
})
if err != nil {
// Handle error (Timeout, QueueFull, or RunnerClosed)
}
Observability
Monitor your worker pool health in real-time.
stats := r.Stats()
fmt.Printf("Active: %d, Queued: %d, Total Processed: %d\n",
stats.ActiveWorkers,
stats.QueuedTasks,
stats.TotalProcessed,
)
Integration with Server
task.Runner implements the standard Start/Stop pattern, making it easy to integrate with application lifecycle managers.
// In your main.go
func run() {
runner := task.NewRunner(task.WithMaxWorkers(50))
srv := server.New() // Your application server
// Dependency Injection
srv.AddService(runner)
srv.Run() // Runner starts with server and stops gracefully with server
}
Options
| Option | Description | Default |
|---|---|---|
WithMaxWorkers(n) |
Max number of concurrent goroutines. | 10 |
WithQueueSize(n) |
Max number of tasks waiting in queue. | 1000 |
WithErrorHandler(fn) |
Custom callback for handling panics. | Log to stdout |
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrQueueFull 表示任务队列已满，无法接受新任务。 // 调用者应该处理此错误（例如：返回 503，或者降级）。 ErrQueueFull = errors.New("task: queue is full") // ErrRunnerClosed 表示 Runner 已经停止或正在停止，不再接受新任务。 ErrRunnerClosed = errors.New("task: runner is closed") )
Functions ¶
Types ¶
type ErrorHandler ¶
ErrorHandler 用于处理任务执行过程中的 Panic 或错误。 ctx: 发生错误时的上下文 p: recover() 捕获的对象 (可能为 nil)
type Option ¶
type Option func(*config)
Option 定义配置修改函数
func WithErrorHandler ¶
func WithErrorHandler(h ErrorHandler) Option
WithErrorHandler 设置 Panic 处理回调。 建议在此处集成 zap/zerolog 等日志库。
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner 是一个并发限制的任务执行器。
type Stats ¶
type Stats struct {
// 配置参数
MaxWorkers int `json:"max_workers"`
QueueSize int `json:"queue_size"`
// 实时状态
ActiveWorkers int64 `json:"active_workers"` // 当前正在执行任务的 Worker 数
QueuedTasks int `json:"queued_tasks"` // 当前排队等待的任务数
// 累积计数 (自启动以来)
TotalProcessed uint64 `json:"total_processed"` // 成功处理的任务数
TotalPanics uint64 `json:"total_panics"` // 发生的 Panic 总数
TotalRefused uint64 `json:"total_refused"` // 因队列满被拒绝的任务数
}
Stats 包含 Runner 的运行时快照
Source Files