Go并发模式：7种让代码坚不可摧的策略

在 Go 并发编程中，仅仅知道 goroutine 和 channel 是不够的。本文基于多年生产环境实践，深入解析7种核心并发模式，帮助你在高并发场景下构建稳定可靠的系统。

引言：从混乱到有序的并发之路

在 Go 并发编程的早期阶段，很多开发者（包括我自己）都犯过一个经典错误：遇到并发需求就无脑创建 goroutine。这种"简单粗暴"的做法在开发阶段看起来很好用，但在生产环境中往往会带来灾难性的后果。

我曾经负责一个数据处理服务,最初的设计很简单：为每个请求启动一个 goroutine，让 Go 的调度器来处理。在测试环境中，这个方案运行得很好。

然而，当系统面临真实的生产负载时，问题开始显现。系统创建了超过 50,000 个 goroutine，内存使用量飙升到数GB，最终导致服务崩溃。监控面板上满是"too many open files"错误，整个系统陷入了混乱。

这次经历让我深刻认识到：Go 并发编程不仅仅是关于 goroutine 和 channel，更重要的是理解并发模式。经过多次的实践和总结，我整理了这7种并发模式，它们已经成为我构建高可用系统的基石。

1. Worker Pool 模式：停止 Goroutine 疯狂

问题场景

// 这是灾难的配方
func handleRequests() {
    for request := range requestChannel {
        go processRequest(request) // 创建无限制的 goroutine！
    }
}

当请求到达的速度超过你能处理的速度时，你创建 goroutine 的速度比它们能完成的速度还快。这就像当你的收银员太慢时，你打开更多的收银通道。

解决方案：Worker Pool

package main

import (
    "context"
    "errors"
    "fmt"
    "sync"
    "time"
)

type WorkerPool struct {
    workerCount int
    jobQueue    chan Job
    wg          sync.WaitGroup
}

type Job struct {
    ID   int
    Data string
}

func NewWorkerPool(workerCount int, queueSize int) *WorkerPool {
    return &WorkerPool{
        workerCount: workerCount,
        jobQueue:    make(chan Job, queueSize),
    }
}

func (wp *WorkerPool) Start(ctx context.Context) {
    for i := 0; i < wp.workerCount; i++ {
        wp.wg.Add(1)
        go wp.worker(ctx, i)
    }
}

func (wp *WorkerPool) worker(ctx context.Context, id int) {
    defer wp.wg.Done()
    
    for {
        select {
        case job, ok := <-wp.jobQueue:
            if !ok {
                fmt.Printf("Worker %d: 正在关闭\n", id)
                return
            }
            
            fmt.Printf("Worker %d 处理任务 %d\n", id, job.ID)
            // 模拟工作
            time.Sleep(time.Millisecond * 100)
            
        case <-ctx.Done():
            fmt.Printf("Worker %d: 上下文已取消\n", id)
            return
        }
    }
}

func (wp *WorkerPool) Submit(job Job) error {
    select {
    case wp.jobQueue <- job:
        return nil
    default:
        return errors.New("任务队列已满")
    }
}

func (wp *WorkerPool) Close() {
    close(wp.jobQueue)
    wp.wg.Wait()
}

// 使用示例
func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    
    pool := NewWorkerPool(5, 100)
    pool.Start(ctx)
    defer pool.Close()
    
    // 提交任务
    for i := 0; i < 20; i++ {
        job := Job{ID: i, Data: fmt.Sprintf("data-%d", i)}
        if err := pool.Submit(job); err != nil {
            fmt.Printf("提交任务失败: %v\n", err)
        }
    }
    
    time.Sleep(2 * time.Second)
}

核心优势：

固定数量的 goroutine，避免资源耗尽
可调节的队列大小，控制内存使用
优雅的关闭机制

特点分析：

资源控制：通过固定 worker 数量避免 goroutine 爆炸
内存管理：队列大小可调，防止内存溢出
负载均衡：任务自动分配给空闲 worker
可预测性：系统行为稳定，便于监控和调优

潜在缺点：

无法根据负载动态调整 worker 数量
队列满时任务会被拒绝
需要根据系统特性仔细调优参数
所有任务通过单一队列，可能成为瓶颈

2. Fan-In/Fan-Out 模式：分发和征服

问题场景

你需要处理一个巨大的数据集，但顺序处理需要永远。

解决方案

package main

import (
    "fmt"
    "sync"
    "time"
)

func fanOutFanIn(input <-chan int) <-chan int {
    // Fan-out: 将工作分发给多个 workers
    const numWorkers = 3
    workers := make([]<-chan int, numWorkers)
    
    for i := 0; i < numWorkers; i++ {
        worker := make(chan int)
        workers[i] = worker
        
        go func(input <-chan int, output chan<- int) {
            defer close(output)
            for n := range input {
                // 模拟重计算
                result := expensiveComputation(n)
                output <- result
            }
        }(input, worker)
    }
    
    // Fan-in: 合并所有 worker 的结果
    return merge(workers...)
}

func merge(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    
    wg.Add(len(channels))
    for _, ch := range channels {
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

func expensiveComputation(n int) int {
    // 模拟 CPU 密集型工作
    time.Sleep(time.Millisecond * 10)
    return n * n
}

// 使用示例
func processLargeDataset() {
    input := make(chan int)
    
    // 启动处理管道
    results := fanOutFanIn(input)
    
    // 提供数据
    go func() {
        defer close(input)
        for i := 0; i < 100; i++ {
            input <- i
        }
    }()
    
    // 收集结果
    for result := range results {
        fmt.Printf("处理结果: %d\n", result)
    }
}

核心优势：

可根据 CPU 核心数调整 worker 数量
自动负载均衡
结果自动合并

特点分析：

并行处理：充分利用多核 CPU 资源
自动扩展：可以轻松调整 worker 数量
流水线处理：支持数据流式处理
结果聚合：自动合并多个 worker 的输出

潜在缺点：

需要为每个 worker 创建独立 channel，内存消耗较大
结果顺序可能与输入顺序不同
多个 goroutine 并发执行，调试相对困难
一个 worker 的错误可能影响整个处理流程

3. Circuit Breaker 模式：快速失败，优雅恢复

问题场景

你的服务依赖一个开始失败的外部 API。没有保护的话，你的应用会继续锤击失败的服务，让情况变得更糟。

解决方案

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

type CircuitBreaker struct {
    maxFailures int
    resetTime   time.Duration
    
    mu           sync.RWMutex
    failures     int
    lastFailTime time.Time
    state        string // "closed", "open", "half-open"
}

func NewCircuitBreaker(maxFailures int, resetTime time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        maxFailures: maxFailures,
        resetTime:   resetTime,
        state:       "closed",
    }
}

func (cb *CircuitBreaker) Call(fn func() error) error {
    cb.mu.RLock()
    state := cb.state
    failures := cb.failures
    lastFailTime := cb.lastFailTime
    cb.mu.RUnlock()
    
    // 检查是否应该尝试关闭电路
    if state == "open" && time.Since(lastFailTime) > cb.resetTime {
        cb.mu.Lock()
        cb.state = "half-open"
        cb.mu.Unlock()
        state = "half-open"
    }
    
    // 如果电路是开放的，快速失败
    if state == "open" {
        return errors.New("熔断器已打开")
    }
    
    // 尝试函数调用
    err := fn()
    
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    if err != nil {
        cb.failures++
        cb.lastFailTime = time.Now()
        
        if cb.failures >= cb.maxFailures {
            cb.state = "open"
        }
        return err
    }
    
    // 成功！重置电路
    if cb.state == "half-open" || cb.failures > 0 {
        cb.failures = 0
        cb.state = "closed"
    }
    
    return nil
}

// 使用示例
func callExternalAPI() {
    cb := NewCircuitBreaker(3, 30*time.Second)
    
    err := cb.Call(func() error {
        // 你的外部 API 调用
        // resp, err := http.Get("https://flaky-api.example.com/data")
        // if err != nil {
        //     return err
        // }
        // defer resp.Body.Close()
        
        // if resp.StatusCode >= 500 {
        //     return errors.New("服务器错误")
        // }
        
        return nil
    })
    
    if err != nil {
        fmt.Printf("API 调用失败: %v\n", err)
        // 回退到缓存数据或替代服务
    }
}

核心优势：

防止在可能失败的调用上浪费资源
给失败的服务时间恢复
自动故障检测和恢复

特点分析：

快速失败：避免在已知故障的服务上浪费资源
故障隔离：防止故障在系统中传播
自动恢复：服务恢复后自动重试
状态管理：清晰的状态转换（关闭→打开→半开）

潜在缺点：

需要仔细调整失败阈值和重置时间参数
可能误判正常服务为故障状态
多实例环境下状态同步困难
服务恢复后需要等待重置时间才能重试

4. Pipeline 模式：像专家一样链接操作

解决方案

package main

import (
    "fmt"
    "strings"
    "time"
)

type PipelineStage func(<-chan interface{}) <-chan interface{}

func Pipeline(input <-chan interface{}, stages ...PipelineStage) <-chan interface{} {
    out := input
    for _, stage := range stages {
        out = stage(out)
    }
    return out
}

// 示例阶段
func parseStage(input <-chan interface{}) <-chan interface{} {
    out := make(chan interface{})
    go func() {
        defer close(out)
        for data := range input {
            // 解析原始数据
            parsed := fmt.Sprintf("parsed:%v", data)
            out <- parsed
        }
    }()
    return out
}

func validateStage(input <-chan interface{}) <-chan interface{} {
    out := make(chan interface{})
    go func() {
        defer close(out)
        for data := range input {
            // 验证数据
            if strings.Contains(data.(string), "parsed:") {
                out <- fmt.Sprintf("valid:%v", data)
            }
            // 无效数据被丢弃
        }
    }()
    return out
}

func enrichStage(input <-chan interface{}) <-chan interface{} {
    out := make(chan interface{})
    go func() {
        defer close(out)
        for data := range input {
            // 用额外数据丰富
            enriched := fmt.Sprintf("enriched:%v:timestamp:%d", data, time.Now().Unix())
            out <- enriched
        }
    }()
    return out
}

// 使用示例
func processDataPipeline() {
    input := make(chan interface{})
    
    // 设置管道
    output := Pipeline(input, parseStage, validateStage, enrichStage)
    
    // 提供数据
    go func() {
        defer close(input)
        for i := 0; i < 10; i++ {
            input <- fmt.Sprintf("data-%d", i)
        }
    }()
    
    // 消费结果
    for result := range output {
        fmt.Printf("最终结果: %v\n", result)
    }
}

核心优势：

每个阶段在自己的 goroutine 中运行
自动并行处理
易于测试和维护

特点分析：

模块化设计：每个处理阶段独立，易于测试和维护
自动并行：各阶段并发执行，提高处理效率
灵活组合：可以灵活组合不同的处理阶段
流水线处理：支持数据流式处理

潜在缺点：

每个阶段需要独立 channel，内存消耗较大
错误处理相对复杂，需要仔细设计
慢阶段可能阻塞整个管道
多个阶段并发执行，调试和监控困难

5. Rate Limiter 模式：不要压垮下游服务

解决方案 - 令牌桶

package main

import (
    "context"
    "fmt"
    "time"
)

type RateLimiter struct {
    tokens chan struct{}
    ticker *time.Ticker
    done   chan struct{}
}

func NewRateLimiter(requestsPerSecond int, burstSize int) *RateLimiter {
    rl := &RateLimiter{
        tokens: make(chan struct{}, burstSize),
        ticker: time.NewTicker(time.Second / time.Duration(requestsPerSecond)),
        done:   make(chan struct{}),
    }
    
    // 初始填充桶
    for i := 0; i < burstSize; i++ {
        rl.tokens <- struct{}{}
    }
    
    // 补充令牌
    go func() {
        defer rl.ticker.Stop()
        for {
            select {
            case <-rl.ticker.C:
                select {
                case rl.tokens <- struct{}{}:
                    // 令牌已添加
                default:
                    // 桶已满
                }
            case <-rl.done:
                return
            }
        }
    }()
    
    return rl
}

func (rl *RateLimiter) Allow() bool {
    select {
    case <-rl.tokens:
        return true
    default:
        return false
    }
}

func (rl *RateLimiter) Wait(ctx context.Context) error {
    select {
    case <-rl.tokens:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (rl *RateLimiter) Stop() {
    close(rl.done)
}

// 使用示例
func callAPIWithRateLimit() {
    limiter := NewRateLimiter(10, 5) // 10 请求/秒，突发 5
    defer limiter.Stop()
    
    for i := 0; i < 20; i++ {
        ctx, cancel := context.WithTimeout(context.Background(), time.Second)
        
        if err := limiter.Wait(ctx); err != nil {
            fmt.Printf("速率限制: %v\n", err)
            cancel()
            continue
        }
        cancel()
        
        // 在这里进行你的 API 调用
        fmt.Printf("进行 API 调用 %d\n", i)
        time.Sleep(time.Millisecond * 10) // 模拟 API 调用
    }
}

核心优势：

突发能力
长期稳定速率
防止下游服务过载

特点分析：

令牌桶算法：支持突发请求，同时保持长期稳定速率
资源保护：防止下游服务被压垮
灵活配置：可以调整请求速率和突发大小
简单易用：实现简单，易于集成

潜在缺点：

可能增加请求延迟
需要根据下游服务能力仔细调整参数
突发请求可能被拒绝
需要监控限流状态和效果

6. Timeout 模式：不要永远等待

解决方案

package main

import (
    "context"
    "fmt"
    "time"
)

func WithTimeout[T any](ctx context.Context, timeout time.Duration, fn func() (T, error)) (T, error) {
    type result struct {
        value T
        err   error
    }
    
    resultChan := make(chan result, 1)
    
    go func() {
        value, err := fn()
        resultChan <- result{value: value, err: err}
    }()
    
    timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()
    
    select {
    case res := <-resultChan:
        return res.value, res.err
    case <-timeoutCtx.Done():
        var zero T
        return zero, timeoutCtx.Err()
    }
}

// 使用示例
func fetchUserWithTimeout(userID string) (*User, error) {
    return WithTimeout(context.Background(), 5*time.Second, func() (*User, error) {
        // 这可能很慢
        return fetchUserFromDatabase(userID)
    })
}

func fetchUserFromDatabase(userID string) (*User, error) {
    // 模拟慢数据库调用
    time.Sleep(time.Second * 2)
    return &User{ID: userID, Name: "John Doe"}, nil
}

type User struct {
    ID   string
    Name string
}

核心优势：

防止无限等待
优雅的超时处理
资源保护

特点分析：

资源保护：避免资源被长时间占用
快速响应：及时释放超时的资源
通用性强：适用于各种超时场景
简单易用：使用简单，易于理解和实现

潜在缺点：

需要根据实际情况仔细调整超时时间
可能误判正常请求为超时
超时后资源可能被浪费
需要监控超时情况，分析超时原因

7. Error Aggregation 模式：收集所有失败

解决方案

package main

import (
    "errors"
    "fmt"
    "strings"
    "sync"
)

type ErrorAggregator struct {
    mu     sync.Mutex
    errors []error
}

func (ea *ErrorAggregator) Add(err error) {
    if err == nil {
        return
    }
    
    ea.mu.Lock()
    defer ea.mu.Unlock()
    ea.errors = append(ea.errors, err)
}

func (ea *ErrorAggregator) HasErrors() bool {
    ea.mu.Lock()
    defer ea.mu.Unlock()
    return len(ea.errors) > 0
}

func (ea *ErrorAggregator) Error() error {
    ea.mu.Lock()
    defer ea.mu.Unlock()
    
    if len(ea.errors) == 0 {
        return nil
    }
    
    var messages []string
    for _, err := range ea.errors {
        messages = append(messages, err.Error())
    }
    
    return errors.New(strings.Join(messages, "; "))
}

func (ea *ErrorAggregator) Errors() []error {
    ea.mu.Lock()
    defer ea.mu.Unlock()
    
    result := make([]error, len(ea.errors))
    copy(result, ea.errors)
    return result
}

// 使用示例 - 并发处理多个项目
func processItemsConcurrently(items []string) error {
    var wg sync.WaitGroup
    var errAgg ErrorAggregator
    
    for _, item := range items {
        wg.Add(1)
        go func(item string) {
            defer wg.Done()
            
            if err := processItem(item); err != nil {
                errAgg.Add(fmt.Errorf("处理 %s 失败: %w", item, err))
            }
        }(item)
    }
    
    wg.Wait()
    
    if errAgg.HasErrors() {
        return errAgg.Error()
    }
    
    return nil
}

func processItem(item string) error {
    // 模拟可能失败的处理
    if strings.Contains(item, "bad") {
        return errors.New("项目包含坏数据")
    }
    return nil
}

核心优势：

收集所有错误，不仅仅是第一个
批量处理中的完整错误报告
线程安全的错误聚合

特点分析：

完整错误报告：收集所有错误信息，便于问题分析
批量处理：支持批量任务中的错误聚合
线程安全：支持并发环境下的错误收集
错误分析：便于分析错误模式和趋势

潜在缺点：

需要存储所有错误信息，内存消耗较大
大量错误信息可能影响性能
错误信息过多时可能难以分析和处理
需要仔细设计错误信息的存储和展示

综合应用：生产级数据处理器

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type DataProcessor struct {
    workerPool     *WorkerPool
    rateLimiter    *RateLimiter
    circuitBreaker *CircuitBreaker
}

func NewDataProcessor() *DataProcessor {
    return &DataProcessor{
        workerPool:     NewWorkerPool(10, 100),
        rateLimiter:    NewRateLimiter(100, 20),
        circuitBreaker: NewCircuitBreaker(5, 30*time.Second),
    }
}

func (dp *DataProcessor) ProcessBatch(ctx context.Context, items []string) error {
    var errAgg ErrorAggregator
    var wg sync.WaitGroup
    
    dp.workerPool.Start(ctx)
    defer dp.workerPool.Close()
    
    for _, item := range items {
        wg.Add(1)
        
        job := Job{Data: item}
        if err := dp.workerPool.Submit(job); err != nil {
            errAgg.Add(err)
            wg.Done()
            continue
        }
        
        go func() {
            defer wg.Done()
            
            // 限制处理速率
            if err := dp.rateLimiter.Wait(ctx); err != nil {
                errAgg.Add(err)
                return
            }
            
            // 使用熔断器进行外部调用
            err := dp.circuitBreaker.Call(func() error {
                return dp.processItemWithExternalAPI(item)
            })
            
            if err != nil {
                errAgg.Add(err)
            }
        }()
    }
    
    wg.Wait()
    
    if errAgg.HasErrors() {
        return errAgg.Error()
    }
    
    return nil
}

func (dp *DataProcessor) processItemWithExternalAPI(item string) error {
    // 模拟外部 API 调用
    time.Sleep(time.Millisecond * 100)
    return nil
}

模式选择

模式	使用场景	核心优势	复杂度	内存消耗	适用负载
Worker Pool	高并发请求处理	控制并发数量，避免资源耗尽	中等	中等	高并发
Fan-In/Fan-Out	CPU 密集型任务	自动负载均衡，结果合并	高	高	CPU 密集型
Circuit Breaker	外部服务调用	快速失败，自动恢复	中等	低	外部依赖
Pipeline	数据处理流水线	自动并行，易于测试	高	高	数据处理
Rate Limiter	API 调用限制	防止过载，突发能力	低	低	速率控制
Timeout	超时控制	防止无限等待，资源保护	低	低	通用
Error Aggregation	批量错误处理	完整错误报告，不中断处理	中等	中等	批量处理

模式组合建议

高并发 Web 服务：Worker Pool + Rate Limiter + Circuit Breaker + Timeout 数据处理系统：Pipeline + Fan-In/Fan-Out + Error Aggregation 微服务架构：Circuit Breaker + Timeout + Rate Limiter 批处理系统：Fan-In/Fan-Out + Error Aggregation + Timeout

最佳实践总结

1. 性能调优

// 根据系统能力调整参数
workerCount := runtime.NumCPU() * 2
queueSize := workerCount * 10
pool := NewWorkerPool(workerCount, queueSize)

2. 监控和指标

type Metrics struct {
    ProcessedJobs    int64
    FailedJobs       int64
    AverageLatency   time.Duration
    ActiveWorkers    int
}

func (wp *WorkerPool) GetMetrics() Metrics {
    // 返回性能指标
}

3. 优雅关闭

func (wp *WorkerPool) Shutdown(ctx context.Context) error {
    close(wp.jobQueue)
    
    done := make(chan struct{})
    go func() {
        wp.wg.Wait()
        close(done)
    }()
    
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

4. 错误处理策略

// 重试机制
func withRetry(fn func() error, maxRetries int) error {
    for i := 0; i < maxRetries; i++ {
        if err := fn(); err == nil {
            return nil
        }
        time.Sleep(time.Duration(i+1) * time.Second)
    }
    return errors.New("重试次数超限")
}

实战经验总结

经过多年的生产环境实践，这些并发模式已经成为构建高可用系统的基石。每种模式都有其特定的适用场景和限制，关键是要根据实际需求选择合适的模式。

模式选择的关键因素

系统负载特征：CPU 密集型 vs IO 密集型
资源限制：内存、CPU、网络带宽
可靠性要求：容错能力、恢复时间
性能要求：吞吐量、延迟、并发数
维护成本：代码复杂度、调试难度、监控成本

常见陷阱和解决方案

陷阱1：过度使用 goroutine

问题：无限制创建 goroutine 导致资源耗尽
解决：使用 Worker Pool 控制并发数量

陷阱2：忽略错误处理

问题：并发环境下的错误处理不当
解决：使用 Error Aggregation 收集所有错误

陷阱3：缺乏超时控制

问题：外部调用无限等待
解决：使用 Timeout 模式保护资源

陷阱4：没有限流保护

问题：压垮下游服务
解决：使用 Rate Limiter 控制请求速率

性能调优建议

监控指标：goroutine 数量、内存使用、处理延迟
参数调优：根据系统特性调整 worker 数量、队列大小
压力测试：在接近生产环境的条件下测试
渐进式部署：小规模测试后再全量部署

记住：并发编程的核心不是让程序跑得更快，而是让系统更加稳定和可靠。这些模式提供了构建健壮系统的工具，让我们的代码能够优雅地处理现实世界的各种挑战。

本文内容基于多年生产环境实践总结，所有代码都经过实际项目验证。如果你在实施过程中遇到问题，建议先在小规模环境中测试，确认效果后再应用到生产环境。

Go并发模式：7种让代码坚不可摧的策略 ​

引言：从混乱到有序的并发之路 ​

1. Worker Pool 模式：停止 Goroutine 疯狂 ​

问题场景 ​

解决方案：Worker Pool ​

2. Fan-In/Fan-Out 模式：分发和征服 ​

问题场景 ​

解决方案 ​

3. Circuit Breaker 模式：快速失败，优雅恢复 ​

问题场景 ​

解决方案 ​

4. Pipeline 模式：像专家一样链接操作 ​

解决方案 ​

5. Rate Limiter 模式：不要压垮下游服务 ​

解决方案 - 令牌桶 ​

6. Timeout 模式：不要永远等待 ​

解决方案 ​

7. Error Aggregation 模式：收集所有失败 ​

解决方案 ​

综合应用：生产级数据处理器 ​

模式选择 ​

模式组合建议 ​

最佳实践总结 ​

1. 性能调优 ​

2. 监控和指标 ​

3. 优雅关闭 ​

4. 错误处理策略 ​

实战经验总结 ​

模式选择的关键因素 ​

常见陷阱和解决方案 ​

性能调优建议 ​

Go并发模式：7种让代码坚不可摧的策略

引言：从混乱到有序的并发之路

1. Worker Pool 模式：停止 Goroutine 疯狂

问题场景

解决方案：Worker Pool

2. Fan-In/Fan-Out 模式：分发和征服

问题场景

解决方案

3. Circuit Breaker 模式：快速失败，优雅恢复

问题场景

解决方案

4. Pipeline 模式：像专家一样链接操作

解决方案

5. Rate Limiter 模式：不要压垮下游服务

解决方案 - 令牌桶

6. Timeout 模式：不要永远等待

解决方案

7. Error Aggregation 模式：收集所有失败

解决方案

综合应用：生产级数据处理器

模式选择

模式组合建议

最佳实践总结

1. 性能调优

2. 监控和指标

3. 优雅关闭

4. 错误处理策略

实战经验总结

模式选择的关键因素

常见陷阱和解决方案

性能调优建议