Golang任务队列与异步处理教程

type Task struct {
    ID      string
    Payload map[string]interface{}
    Process func(map[string]interface{}) error
}

const QueueSize = 100
const WorkerCount = 5
var TaskQueue = make(chan Task, QueueSize)
func StartWorkers() {
for i := 0; i < WorkerCount; i++ {
go func(workerID int) {
for task := range TaskQueue {
println("Worker", workerID, "processing task:", task.ID)
err := task.Process(task.Payload)
if err != nil {
println("Task failed:", task.ID, "Error:", err.Error())
} else {
println("Task completed:", task.ID)
}
}
}(i)
}
}

func SendEmailAsync(to, subject, body string) {
    task := Task{
        ID:      fmt.Sprintf("email-%d", time.Now().UnixNano()),
        Payload: map[string]interface{}{"to": to, "subject": subject, "body": body},
        Process: func(data map[string]interface{}) error {
            // 模拟耗时操作
            time.Sleep(2 * time.Second)
            fmt.Printf("? Sending email to %v: %v\n", data["to"], data["subject"])
            return nil
        },
    }
    TaskQueue 
调用该函数不会阻塞主流程：
SendEmailAsync("user@example.com", "Welcome!", "Thank you for signing up.")
println("Email task submitted, continuing...")
3. 集成Redis实现持久化队列（可选）
对于生产环境，建议使用Redis等消息中间件保证任务不丢失。
使用go-redis/redis库连接Redis，将任务序列化为JSON存入List结构：
rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
func PushTaskToRedis(task Task) error {
data, _ := json.Marshal(task)
return rdb.LPush("task_queue", data).Err()
}
Worker从Redis轮询获取任务：
func RedisWorker() {
    for {
        val, err := rdb.BLPop(0, "task_queue").Result()
        if err != nil && err != redis.Nil {
            log.Println("Redis error:", err)
            continue
        }
        if len(val) == 2 {
            var task Task
            json.Unmarshal([]byte(val[1]), &task)
            task.Process(task.Payload)
        }
    }
}
4. 错误处理与重试机制
异步任务需考虑失败场景，加入基本重试逻辑：
type Task struct {
    ID       string
    Payload  map[string]interface{}
    MaxRetry int
    Retried  int
    Process  func(map[string]interface{}) error
}
// 在worker中判断重试次数
if err := task.Process(task.Payload); err != nil {
if task.Retried < task.MaxRetry {
task.Retried++
TaskQueue <- task // 重新入队
} else {
println("Task permanently failed:", task.ID)
}
}
基本上就这些。这个示例展示了Go中实现任务队列的核心思路：使用channel做内存队列，配合goroutine处理异步任务，再根据需求扩展持久化和错误恢复能力。不复杂但容易忽略细节，比如channel容量设置和worker退出机制。实际项目中可考虑使用成熟框架如Machinery或Asynq来降低维护成本。
今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~