Golang 作业管理插件 job 的使用

在 Golang 中，将短期命令转换为长期任务并管理这些作业可以使用一些优秀的作业管理库。下面我将介绍如何使用 github.com/albrow/jobs 这个轻量级的作业队列库来实现这个功能。

安装 jobs 库

首先安装 jobs 库：

go get github.com/albrow/jobs

基本使用方法

1. 初始化作业队列

package main

import (
	"fmt"
	"time"
	
	"github.com/albrow/jobs"
)

func main() {
	// 初始化作业队列
	jobQueue, err := jobs.NewQueue(&jobs.QueueConfig{
		Name:                 "default",
		NumWorkers:           4,  // 工作线程数
		WorkerWaitTime:       time.Second, // 工作线程等待时间
		PollInterval:         time.Second, // 轮询间隔
		ReserveJobBufferSize: 10, // 预留作业缓冲区大小
	})
	if err != nil {
		panic(err)
	}
	
	// 注册作业处理函数
	jobQueue.Register("example_job", exampleJobHandler)
	
	// 启动队列
	jobQueue.Start()
	
	// 添加作业到队列
	jobID, err := jobQueue.Enqueue("example_job", map[string]interface{}{
		"param1": "value1",
		"param2": 123,
	})
	if err != nil {
		panic(err)
	}
	
	fmt.Printf("Job enqueued with ID: %s\n", jobID)
	
	// 等待作业完成
	time.Sleep(5 * time.Second)
	
	// 停止队列
	jobQueue.Stop()
}

// 作业处理函数
func exampleJobHandler(params map[string]interface{}) error {
	fmt.Println("Processing job with params:", params)
	// 模拟长时间运行的任务
	time.Sleep(2 * time.Second)
	return nil
}

2. 更高级的作业管理

package main

import (
	"fmt"
	"log"
	"time"
	
	"github.com/albrow/jobs"
)

func main() {
	// 初始化队列
	queue, err := jobs.NewQueue(&jobs.QueueConfig{
		Name:       "worker_queue",
		NumWorkers: 5,
	})
	if err != nil {
		log.Fatal(err)
	}
	
	// 注册多个作业类型
	queue.Register("email_job", sendEmailHandler)
	queue.Register("report_job", generateReportHandler)
	queue.Register("cleanup_job", cleanupHandler)
	
	// 启动队列
	queue.Start()
	defer queue.Stop() // 确保程序退出时停止队列
	
	// 添加多个作业
	for i := 0; i < 10; i++ {
		_, err := queue.Enqueue("email_job", map[string]interface{}{
			"to":      fmt.Sprintf("user%d@example.com", i),
			"subject": "Test Email",
			"body":    "This is a test email",
		})
		if err != nil {
			log.Printf("Failed to enqueue email job: %v", err)
		}
	}
	
	// 添加定期作业
	ticker := time.NewTicker(1 * time.Hour)
	go func() {
		for range ticker.C {
			_, err := queue.Enqueue("cleanup_job", nil)
			if err != nil {
				log.Printf("Failed to enqueue cleanup job: %v", err)
			}
		}
	}()
	
	// 保持程序运行
	select {}
}

func sendEmailHandler(params map[string]interface{}) error {
	to := params["to"].(string)
	subject := params["subject"].(string)
	body := params["body"].(string)
	
	// 模拟发送电子邮件
	log.Printf("Sending email to %s: %s - %s", to, subject, body)
	time.Sleep(500 * time.Millisecond)
	return nil
}

func generateReportHandler(params map[string]interface{}) error {
	// 模拟生成报告
	log.Println("Generating report...")
	time.Sleep(2 * time.Second)
	return nil
}

func cleanupHandler(params map[string]interface{}) error {
	// 模拟清理任务
	log.Println("Performing system cleanup...")
	time.Sleep(1 * time.Second)
	return nil
}

作业持久化

默认情况下，jobs 库使用内存存储作业。如果需要持久化，可以配置 Redis 作为后端：

package main

import (
	"fmt"
	"time"
	
	"github.com/albrow/jobs"
	"github.com/gomodule/redigo/redis"
)

func main() {
	// 创建 Redis 连接池
	redisPool := &redis.Pool{
		MaxIdle:     3,
		IdleTimeout: 240 * time.Second,
		Dial: func() (redis.Conn, error) {
			return redis.Dial("tcp", "localhost:6379")
		},
	}
	
	// 使用 Redis 作为后端初始化队列
	jobQueue, err := jobs.NewQueue(&jobs.QueueConfig{
		Name:       "persistent_queue",
		NumWorkers: 4,
		Backend:    jobs.NewRedisBackend(redisPool),
	})
	if err != nil {
		panic(err)
	}
	
	// ... 其余代码与之前相同 ...
}

作业状态监控

// 获取作业状态
status, err := jobQueue.JobStatus(jobID)
if err != nil {
	log.Printf("Error getting job status: %v", err)
} else {
	log.Printf("Job status: %s", status)
}

// 获取队列统计信息
stats := jobQueue.Stats()
log.Printf("Queue stats: %+v", stats)

作业重试机制

func retryableJobHandler(params map[string]interface{}) error {
	// 模拟可能失败的任务
	if time.Now().Unix()%2 == 0 {
		return fmt.Errorf("simulated error")
	}
	return nil
}

// 注册作业时指定重试策略
jobQueue.RegisterWithOptions("retry_job", retryableJobHandler, &jobs.HandlerOptions{
	MaxRetries: 3,
	RetryDelay: 5 * time.Second,
})

jobs 库提供了一种简单而有效的方式来管理 Golang 中的长期任务。它支持：

• 多工作线程并行处理
• 作业持久化（通过 Redis）
• 作业状态跟踪
• 重试机制
• 定期作业

对于更复杂的场景，你也可以考虑使用其他更成熟的作业队列系统如 RabbitMQ、Kafka 或 Google Cloud Tasks 等。