Golang取消并发任务：context与channel实战指南

在Golang中实现可取消的并发任务，核心在于利用context.Context和channel的组合。1. context负责传递取消信号；2. channel用于任务间的通信和结果传递；3. 通过sync.WaitGroup确保所有任务完成后再关闭通道；4. 使用select监听ctx.Done()以响应取消信号；5. 结合time.After实现任务超时控制；6. 利用defer和recover捕获并处理任务中的panic；7. 通过worker pool模式优化大量任务的并发执行效率。这种方案不仅支持灵活的任务生命周期管理，还提供了优雅关闭和错误处理机制，从而提升程序的健壮性和并发性能。

Golang中实现可取消的并发任务，核心在于利用context.Context和channel的组合。context负责传递取消信号，channel则用于任务间的通信和结果传递。这种方式既能保证并发执行的效率，又能灵活地控制任务的生命周期。

解决方案

package main

import (
    "context"
    "fmt"
    "math/rand"
    "sync"
    "time"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // 确保在main函数退出时取消所有任务

    numTasks := 5
    results := make(chan int, numTasks)
    errChan := make(chan error, numTasks) // 用于收集错误
    var wg sync.WaitGroup

    for i := 0; i < numTasks; i++ {
        wg.Add(1)
        go func(taskID int) {
            defer wg.Done()
            select {
            case <-ctx.Done():
                fmt.Printf("任务 %d 被取消\n", taskID)
                return // 任务被取消，直接返回
            case res, err := performTask(ctx, taskID): // 传递context
                if err != nil {
                    errChan <- fmt.Errorf("任务 %d 失败: %w", taskID, err)
                    return
                }
                results <- res
            }
        }(i)
    }

    // 模拟取消操作，在一定时间后取消任务
    time.AfterFunc(2*time.Second, func() {
        fmt.Println("取消所有任务...")
        cancel()
    })

    go func() {
        wg.Wait()
        close(results) // 关闭结果通道
        close(errChan) // 关闭错误通道
    }()

    // 处理结果和错误
    for res := range results {
        fmt.Printf("任务结果: %d\n", res)
    }

    for err := range errChan {
        fmt.Println("错误:", err)
    }

    fmt.Println("程序结束")
}

func performTask(ctx context.Context, taskID int) (int, error) {
    // 模拟耗时操作
    delay := time.Duration(rand.Intn(5)) * time.Second
    fmt.Printf("任务 %d 开始，预计耗时 %v\n", taskID, delay)

    select {
    case <-time.After(delay):
        // 模拟任务完成
        result := taskID * 10
        fmt.Printf("任务 %d 完成，结果: %d\n", taskID, result)
        return result, nil
    case <-ctx.Done():
        // 任务被取消
        fmt.Printf("任务 %d 被取消\n", taskID)
        return 0, fmt.Errorf("任务 %d 被取消", taskID)
    }
}

如何优雅地处理任务超时？

任务超时是并发编程中常见的问题。除了使用context.WithTimeout，还可以结合select语句和time.After通道，实现更灵活的超时控制。例如，可以在performTask函数中使用select同时监听ctx.Done()和time.After()，一旦超时或接收到取消信号，立即退出任务。

func performTask(ctx context.Context, taskID int) (int, error) {
    delay := time.Duration(rand.Intn(5)) * time.Second
    fmt.Printf("任务 %d 开始，预计耗时 %v\n", taskID, delay)

    timeout := 3 * time.Second // 设置超时时间
    select {
    case <-time.After(delay):
        result := taskID * 10
        fmt.Printf("任务 %d 完成，结果: %d\n", taskID, result)
        return result, nil
    case <-ctx.Done():
        fmt.Printf("任务 %d 被取消\n", taskID)
        return 0, fmt.Errorf("任务 %d 被取消", taskID)
    case <-time.After(timeout): // 添加超时处理
        fmt.Printf("任务 %d 超时\n", taskID)
        return 0, fmt.Errorf("任务 %d 超时", taskID)
    }
}

如何处理任务执行过程中出现的panic？

在并发任务中，panic会导致程序崩溃。为了避免这种情况，可以使用recover函数捕获panic，并记录错误信息。可以将recover放在goroutine的defer语句中，确保在任务结束时执行。

func performTask(ctx context.Context, taskID int) (int, error) {
    defer func() {
        if r := recover(); r != nil {
            fmt.Printf("任务 %d 发生panic: %v\n", taskID, r)
            // 可以将panic信息发送到错误通道
            // errChan <- fmt.Errorf("任务 %d 发生panic: %v", taskID, r)
        }
    }()

    delay := time.Duration(rand.Intn(5)) * time.Second
    fmt.Printf("任务 %d 开始，预计耗时 %v\n", taskID, delay)

    select {
    case <-time.After(delay):
        // 模拟可能发生panic的情况
        if rand.Intn(10) == 0 {
            panic("模拟panic")
        }
        result := taskID * 10
        fmt.Printf("任务 %d 完成，结果: %d\n", taskID, result)
        return result, nil
    case <-ctx.Done():
        fmt.Printf("任务 %d 被取消\n", taskID)
        return 0, fmt.Errorf("任务 %d 被取消", taskID)
    }
}

如何实现任务的优雅关闭？

优雅关闭指的是在程序退出时，等待所有正在执行的任务完成，而不是强制终止。这可以通过sync.WaitGroup和context.Context来实现。在启动任务时，wg.Add(1)，在任务完成或被取消时，wg.Done()。在程序退出前，调用wg.Wait()等待所有任务完成。同时，使用context.Context传递取消信号，让任务有机会清理资源。

在main函数中，已经使用了sync.WaitGroup，确保所有任务都完成后再关闭通道。取消信号通过context传递，任务内部通过select监听取消信号，实现优雅关闭。

如何使用更高级的并发控制模式，比如worker pool？

对于需要处理大量任务的情况，使用worker pool可以更有效地利用系统资源。worker pool维护一组worker goroutine，从任务队列中获取任务并执行。这可以避免频繁创建和销毁goroutine，提高性能。

package main

import (
    "context"
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// 任务结构体
type Task struct {
    ID int
}

func main() {
    numWorkers := 3
    numTasks := 10

    taskQueue := make(chan Task, numTasks)
    results := make(chan int, numTasks)
    errChan := make(chan error, numTasks)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    var wg sync.WaitGroup

    // 启动worker goroutine
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(ctx, i, taskQueue, results, errChan, &wg)
    }

    // 提交任务
    for i := 0; i < numTasks; i++ {
        taskQueue <- Task{ID: i}
    }
    close(taskQueue) // 关闭任务队列

    // 模拟取消操作
    time.AfterFunc(2*time.Second, func() {
        fmt.Println("取消所有任务...")
        cancel()
    })

    go func() {
        wg.Wait()
        close(results)
        close(errChan)
    }()

    // 处理结果和错误
    for res := range results {
        fmt.Printf("任务结果: %d\n", res)
    }

    for err := range errChan {
        fmt.Println("错误:", err)
    }

    fmt.Println("程序结束")
}

// worker goroutine
func worker(ctx context.Context, workerID int, taskQueue <-chan Task, results chan<- int, errChan chan<- error, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range taskQueue {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d 停止工作\n", workerID)
            return
        default:
            res, err := performTaskWithID(ctx, task.ID, workerID)
            if err != nil {
                errChan <- fmt.Errorf("Worker %d 执行任务 %d 失败: %w", workerID, task.ID, err)
                continue
            }
            results <- res
        }
    }
    fmt.Printf("Worker %d 退出\n", workerID)
}

func performTaskWithID(ctx context.Context, taskID int, workerID int) (int, error) {
    delay := time.Duration(rand.Intn(5)) * time.Second
    fmt.Printf("Worker %d 开始执行任务 %d，预计耗时 %v\n", workerID, taskID, delay)

    select {
    case <-time.After(delay):
        result := taskID * 10
        fmt.Printf("Worker %d 完成任务 %d，结果: %d\n", workerID, taskID, result)
        return result, nil
    case <-ctx.Done():
        fmt.Printf("Worker %d 取消任务 %d\n", workerID, taskID)
        return 0, fmt.Errorf("Worker %d 取消任务 %d", workerID, taskID)
    }
}