mindmap
  root((Go并发设计原则))
    (通信优于共享)
      Channel通信
      消息传递
      避免共享状态
    (简单性原则)
      单一职责
      清晰接口
      最小复杂度
    (可组合性)
      模块化设计
      接口抽象
      组件复用
    (错误处理)
      优雅降级
      超时控制
      资源清理
    (性能考量)
      避免过度并发
      合理的goroutine数量
      内存使用优化

// 并发设计决策框架
type ConcurrencyDesignFramework struct {
    // 问题分析
    ProblemAnalysis struct {
        // 是否需要并发？
        NeedsConcurrency bool
        // 并发类型：CPU密集 vs I/O密集
        ConcurrencyType string
        // 数据依赖关系
        DataDependencies []string
        // 性能要求
        PerformanceRequirements struct {
            Latency    time.Duration
            Throughput int
            Memory     int64
        }
    }
    
    // 工具选择
    ToolSelection struct {
        // 通信方式：Channel vs 共享内存
        CommunicationMethod string
        // 同步原语选择
        SyncPrimitives []string
        // 并发模式
        ConcurrencyPatterns []string
    }
    
    // 实现策略
    ImplementationStrategy struct {
        // Goroutine数量控制
        GoroutinePoolSize int
        // 缓冲区大小
        BufferSizes map[string]int
        // 超时设置
        Timeouts map[string]time.Duration
        // 错误处理策略
        ErrorHandling string
    }
}

// 决策辅助函数
func ChooseConcurrencyApproach(problem ProblemDescription) ConcurrencyApproach {
    approach := ConcurrencyApproach{}
    
    // 1. 分析问题特征
    if problem.IsIOIntensive {
        approach.PreferredPattern = "ProducerConsumer"
        approach.CommunicationMethod = "Channel"
    } else if problem.IsCPUIntensive {
        approach.PreferredPattern = "WorkerPool"
        approach.CommunicationMethod = "SharedMemory"
    }
    
    // 2. 考虑数据流向
    if problem.HasComplexDataFlow {
        approach.PreferredPattern = "Pipeline"
        approach.CommunicationMethod = "Channel"
    }
    
    // 3. 考虑错误处理需求
    if problem.RequiresRobustErrorHandling {
        approach.ErrorStrategy = "ContextCancellation"
        approach.TimeoutStrategy = "Hierarchical"
    }
    
    return approach
}

type ProblemDescription struct {
    IsIOIntensive                bool
    IsCPUIntensive              bool
    HasComplexDataFlow          bool
    RequiresRobustErrorHandling bool
    ExpectedLoad                int
    LatencyRequirement          time.Duration
}

type ConcurrencyApproach struct {
    PreferredPattern      string
    CommunicationMethod   string
    ErrorStrategy         string
    TimeoutStrategy       string
}

// 高性能生产者-消费者实现
type ProducerConsumer struct {
    // 配置参数
    config ProducerConsumerConfig
    
    // 通信通道
    taskQueue   chan Task
    resultQueue chan Result
    errorQueue  chan error
    
    // 控制通道
    ctx    context.Context
    cancel context.CancelFunc
    
    // 同步原语
    wg sync.WaitGroup
    
    // 监控指标
    metrics *ProducerConsumerMetrics
}

type ProducerConsumerConfig struct {
    // 队列配置
    TaskQueueSize   int
    ResultQueueSize int
    ErrorQueueSize  int
    
    // 工作者配置
    ProducerCount int
    ConsumerCount int
    
    // 性能配置
    BatchSize      int
    FlushInterval  time.Duration
    MaxRetries     int
    RetryDelay     time.Duration
    
    // 超时配置
    ProduceTimeout time.Duration
    ConsumeTimeout time.Duration
    ShutdownTimeout time.Duration
}

type Task struct {
    ID       string
    Data     interface{}
    Priority int
    Deadline time.Time
    Retries  int
}

type Result struct {
    TaskID    string
    Data      interface{}
    ProcessTime time.Duration
    Error     error
}

type ProducerConsumerMetrics struct {
    TasksProduced    int64
    TasksConsumed    int64
    TasksCompleted   int64
    TasksFailed      int64
    AverageLatency   time.Duration
    QueueLength      int64
    ActiveConsumers  int32
    ActiveProducers  int32
}

// NewProducerConsumer 创建生产者-消费者系统
func NewProducerConsumer(config ProducerConsumerConfig) *ProducerConsumer {
    ctx, cancel := context.WithCancel(context.Background())
    
    return &ProducerConsumer{
        config:      config,
        taskQueue:   make(chan Task, config.TaskQueueSize),
        resultQueue: make(chan Result, config.ResultQueueSize),
        errorQueue:  make(chan error, config.ErrorQueueSize),
        ctx:         ctx,
        cancel:      cancel,
        metrics:     &ProducerConsumerMetrics{},
    }
}

// Start 启动生产者-消费者系统
func (pc *ProducerConsumer) Start() error {
    // 启动消费者
    for i := 0; i < pc.config.ConsumerCount; i++ {
        pc.wg.Add(1)
        go pc.consumer(i)
    }
    
    // 启动监控goroutine
    pc.wg.Add(1)
    go pc.monitor()
    
    return nil
}

// Produce 生产任务（支持批量和超时）
func (pc *ProducerConsumer) Produce(tasks []Task) error {
    atomic.AddInt32(&pc.metrics.ActiveProducers, 1)
    defer atomic.AddInt32(&pc.metrics.ActiveProducers, -1)
    
    for _, task := range tasks {
        select {
        case pc.taskQueue <- task:
            atomic.AddInt64(&pc.metrics.TasksProduced, 1)
            
        case <-time.After(pc.config.ProduceTimeout):
            return fmt.Errorf("produce timeout for task %s", task.ID)
            
        case <-pc.ctx.Done():
            return pc.ctx.Err()
        }
    }
    
    return nil
}

// ProduceBatch 批量生产（优化版本）
func (pc *ProducerConsumer) ProduceBatch(tasks []Task) error {
    atomic.AddInt32(&pc.metrics.ActiveProducers, 1)
    defer atomic.AddInt32(&pc.metrics.ActiveProducers, -1)
    
    // 分批处理，避免长时间阻塞
    batchSize := pc.config.BatchSize
    for i := 0; i < len(tasks); i += batchSize {
        end := i + batchSize
        if end > len(tasks) {
            end = len(tasks)
        }
        
        batch := tasks[i:end]
        if err := pc.produceBatchInternal(batch); err != nil {
            return err
        }
    }
    
    return nil
}

func (pc *ProducerConsumer) produceBatchInternal(batch []Task) error {
    timeout := time.NewTimer(pc.config.ProduceTimeout)
    defer timeout.Stop()
    
    for _, task := range batch {
        select {
        case pc.taskQueue <- task:
            atomic.AddInt64(&pc.metrics.TasksProduced, 1)
            
        case <-timeout.C:
            return fmt.Errorf("batch produce timeout")
            
        case <-pc.ctx.Done():
            return pc.ctx.Err()
        }
    }
    
    return nil
}

// consumer 消费者goroutine
func (pc *ProducerConsumer) consumer(consumerID int) {
    defer pc.wg.Done()
    
    atomic.AddInt32(&pc.metrics.ActiveConsumers, 1)
    defer atomic.AddInt32(&pc.metrics.ActiveConsumers, -1)
    
    log.Printf("Consumer %d started", consumerID)
    defer log.Printf("Consumer %d stopped", consumerID)
    
    for {
        select {
        case task, ok := <-pc.taskQueue:
            if !ok {
                return // 任务队列已关闭
            }
            
            pc.processTask(consumerID, task)
            
        case <-pc.ctx.Done():
            return
        }
    }
}

// processTask 处理单个任务
func (pc *ProducerConsumer) processTask(consumerID int, task Task) {
    startTime := time.Now()
    atomic.AddInt64(&pc.metrics.TasksConsumed, 1)
    
    defer func() {
        if r := recover(); r != nil {
            log.Printf("Consumer %d panic processing task %s: %v", 
                consumerID, task.ID, r)
            
            pc.sendError(fmt.Errorf("panic in consumer %d: %v", consumerID, r))
            atomic.AddInt64(&pc.metrics.TasksFailed, 1)
        }
    }()
    
    // 检查任务是否过期
    if !task.Deadline.IsZero() && time.Now().After(task.Deadline) {
        log.Printf("Task %s expired", task.ID)
        atomic.AddInt64(&pc.metrics.TasksFailed, 1)
        return
    }
    
    // 处理任务（这里是示例逻辑）
    result, err := pc.executeTask(task)
    
    processTime := time.Since(startTime)
    
    // 发送结果
    select {
    case pc.resultQueue <- Result{
        TaskID:      task.ID,
        Data:        result,
        ProcessTime: processTime,
        Error:       err,
    }:
        if err != nil {
            atomic.AddInt64(&pc.metrics.TasksFailed, 1)
        } else {
            atomic.AddInt64(&pc.metrics.TasksCompleted, 1)
        }
        
    case <-time.After(pc.config.ConsumeTimeout):
        log.Printf("Failed to send result for task %s: timeout", task.ID)
        
    case <-pc.ctx.Done():
        return
    }
    
    // 更新平均延迟
    pc.updateAverageLatency(processTime)
}

// executeTask 执行具体任务逻辑
func (pc *ProducerConsumer) executeTask(task Task) (interface{}, error) {
    // 模拟任务处理
    switch data := task.Data.(type) {
    case string:
        // 字符串处理任务
        time.Sleep(100 * time.Millisecond)
        return strings.ToUpper(data), nil
        
    case int:
        // 数值计算任务
        time.Sleep(50 * time.Millisecond)
        return data * data, nil
        
    default:
        return nil, fmt.Errorf("unsupported task data type: %T", data)
    }
}

// GetResults 获取处理结果
func (pc *ProducerConsumer) GetResults() <-chan Result {
    return pc.resultQueue
}

// GetErrors 获取错误信息
func (pc *ProducerConsumer) GetErrors() <-chan error {
    return pc.errorQueue
}

// sendError 发送错误信息
func (pc *ProducerConsumer) sendError(err error) {
    select {
    case pc.errorQueue <- err:
    default:
        log.Printf("Error queue full, dropping error: %v", err)
    }
}

// updateAverageLatency 更新平均延迟
func (pc *ProducerConsumer) updateAverageLatency(latency time.Duration) {
    // 使用简单的移动平均算法
    currentAvg := atomic.LoadInt64((*int64)(&pc.metrics.AverageLatency))
    newAvg := (time.Duration(currentAvg)*9 + latency) / 10
    atomic.StoreInt64((*int64)(&pc.metrics.AverageLatency), int64(newAvg))
}

// monitor 监控goroutine
func (pc *ProducerConsumer) monitor() {
    defer pc.wg.Done()
    
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            pc.logMetrics()
            
        case <-pc.ctx.Done():
            return
        }
    }
}

// logMetrics 记录监控指标
func (pc *ProducerConsumer) logMetrics() {
    metrics := pc.GetMetrics()
    
    log.Printf("ProducerConsumer Metrics: "+
        "Produced=%d, Consumed=%d, Completed=%d, Failed=%d, "+
        "QueueLen=%d, AvgLatency=%v, ActiveConsumers=%d, ActiveProducers=%d",
        metrics.TasksProduced, metrics.TasksConsumed, metrics.TasksCompleted,
        metrics.TasksFailed, metrics.QueueLength, metrics.AverageLatency,
        metrics.ActiveConsumers, metrics.ActiveProducers)
}

// GetMetrics 获取监控指标
func (pc *ProducerConsumer) GetMetrics() ProducerConsumerMetrics {
    return ProducerConsumerMetrics{
        TasksProduced:   atomic.LoadInt64(&pc.metrics.TasksProduced),
        TasksConsumed:   atomic.LoadInt64(&pc.metrics.TasksConsumed),
        TasksCompleted:  atomic.LoadInt64(&pc.metrics.TasksCompleted),
        TasksFailed:     atomic.LoadInt64(&pc.metrics.TasksFailed),
        AverageLatency:  time.Duration(atomic.LoadInt64((*int64)(&pc.metrics.AverageLatency))),
        QueueLength:     int64(len(pc.taskQueue)),
        ActiveConsumers: atomic.LoadInt32(&pc.metrics.ActiveConsumers),
        ActiveProducers: atomic.LoadInt32(&pc.metrics.ActiveProducers),
    }
}

// Stop 停止生产者-消费者系统
func (pc *ProducerConsumer) Stop() error {
    log.Println("Stopping ProducerConsumer...")
    
    // 关闭任务队列，不再接受新任务
    close(pc.taskQueue)
    
    // 等待所有消费者完成
    done := make(chan struct{})
    go func() {
        pc.wg.Wait()
        close(done)
    }()
    
    select {
    case <-done:
        log.Println("ProducerConsumer stopped gracefully")
        
    case <-time.After(pc.config.ShutdownTimeout):
        log.Println("ProducerConsumer shutdown timeout, forcing stop")
        pc.cancel()
        
        // 再等待一小段时间
        select {
        case <-done:
        case <-time.After(1 * time.Second):
            log.Println("Some goroutines may not have stopped")
        }
    }
    
    // 关闭结果和错误通道
    close(pc.resultQueue)
    close(pc.errorQueue)
    
    return nil
}

// 高性能工作池实现
type WorkerPool struct {
    // 配置
    config WorkerPoolConfig
    
    // 任务分发
    taskQueue chan Job
    
    // 工作者管理
    workers    []*Worker
    workerWG   sync.WaitGroup
    
    // 控制
    ctx    context.Context
    cancel context.CancelFunc
    
    // 监控
    metrics *WorkerPoolMetrics
    
    // 动态调整
    scalingMu sync.RWMutex
    scaling   bool
}

type WorkerPoolConfig struct {
    // 基础配置
    MinWorkers      int
    MaxWorkers      int
    QueueSize       int
    
    // 性能配置
    WorkerTimeout   time.Duration
    IdleTimeout     time.Duration
    ScaleInterval   time.Duration
    
    // 负载均衡
    LoadBalancing   string // "round_robin", "least_loaded", "random"
    
    // 监控配置
    MetricsInterval time.Duration
}

type Job struct {
    ID       string
    Task     func() (interface{}, error)
    Priority int
    Timeout  time.Duration
    Callback func(interface{}, error)
}

type Worker struct {
    id          int
    pool        *WorkerPool
    taskQueue   chan Job
    quit        chan struct{}
    
    // 统计信息
    tasksProcessed int64
    lastActive     time.Time
    
    // 状态
    busy int32
}

type WorkerPoolMetrics struct {
    ActiveWorkers    int32
    IdleWorkers      int32
    QueueLength      int64
    TasksProcessed   int64
    TasksCompleted   int64
    TasksFailed      int64
    AverageLatency   time.Duration
    ThroughputPerSec float64
}

// NewWorkerPool 创建工作池
func NewWorkerPool(config WorkerPoolConfig) *WorkerPool {
    ctx, cancel := context.WithCancel(context.Background())
    
    wp := &WorkerPool{
        config:    config,
        taskQueue: make(chan Job, config.QueueSize),
        workers:   make([]*Worker, 0, config.MaxWorkers),
        ctx:       ctx,
        cancel:    cancel,
        metrics:   &WorkerPoolMetrics{},
    }
    
    return wp
}

// Start 启动工作池
func (wp *WorkerPool) Start() error {
    // 启动最小数量的工作者
    for i := 0; i < wp.config.MinWorkers; i++ {
        wp.addWorker()
    }
    
    // 启动任务分发器
    wp.workerWG.Add(1)
    go wp.dispatcher()
    
    // 启动监控和自动缩放
    wp.workerWG.Add(1)
    go wp.monitor()
    
    if wp.config.MaxWorkers > wp.config.MinWorkers {
        wp.workerWG.Add(1)
        go wp.autoScale()
    }
    
    return nil
}

// Submit 提交任务
func (wp *WorkerPool) Submit(job Job) error {
    select {
    case wp.taskQueue <- job:
        atomic.AddInt64(&wp.metrics.TasksProcessed, 1)
        return nil
        
    case <-wp.ctx.Done():
        return wp.ctx.Err()
        
    default:
        return fmt.Errorf("task queue is full")
    }
}

// SubmitWithTimeout 带超时的任务提交
func (wp *WorkerPool) SubmitWithTimeout(job Job, timeout time.Duration) error {
    select {
    case wp.taskQueue <- job:
        atomic.AddInt64(&wp.metrics.TasksProcessed, 1)
        return nil
        
    case <-time.After(timeout):
        return fmt.Errorf("submit timeout")
        
    case <-wp.ctx.Done():
        return wp.ctx.Err()
    }
}

// addWorker 添加工作者
func (wp *WorkerPool) addWorker() {
    wp.scalingMu.Lock()
    defer wp.scalingMu.Unlock()
    
    if len(wp.workers) >= wp.config.MaxWorkers {
        return
    }
    
    worker := &Worker{
        id:         len(wp.workers),
        pool:       wp,
        taskQueue:  make(chan Job, 1),
        quit:       make(chan struct{}),
        lastActive: time.Now(),
    }
    
    wp.workers = append(wp.workers, worker)
    
    wp.workerWG.Add(1)
    go worker.start()
    
    atomic.AddInt32(&wp.metrics.IdleWorkers, 1)
    
    log.Printf("Added worker %d, total workers: %d", worker.id, len(wp.workers))
}

// removeWorker 移除工作者
func (wp *WorkerPool) removeWorker() {
    wp.scalingMu.Lock()
    defer wp.scalingMu.Unlock()
    
    if len(wp.workers) <= wp.config.MinWorkers {
        return
    }
    
    // 找到最空闲的工作者
    var idleWorker *Worker
    oldestIdle := time.Now()
    
    for _, worker := range wp.workers {
        if atomic.LoadInt32(&worker.busy) == 0 {
            if worker.lastActive.Before(oldestIdle) {
                oldestIdle = worker.lastActive
                idleWorker = worker
            }
        }
    }
    
    if idleWorker != nil {
        close(idleWorker.quit)
        
        // 从切片中移除
        for i, w := range wp.workers {
            if w == idleWorker {
                wp.workers = append(wp.workers[:i], wp.workers[i+1:]...)
                break
            }
        }
        
        atomic.AddInt32(&wp.metrics.IdleWorkers, -1)
        
        log.Printf("Removed worker %d, total workers: %d", 
            idleWorker.id, len(wp.workers))
    }
}

// dispatcher 任务分发器
func (wp *WorkerPool) dispatcher() {
    defer wp.workerWG.Done()
    
    for {
        select {
        case job := <-wp.taskQueue:
            wp.dispatchJob(job)
            
        case <-wp.ctx.Done():
            return
        }
    }
}

// dispatchJob 分发任务到工作者
func (wp *WorkerPool) dispatchJob(job Job) {
    var selectedWorker *Worker
    
    wp.scalingMu.RLock()
    workers := make([]*Worker, len(wp.workers))
    copy(workers, wp.workers)
    wp.scalingMu.RUnlock()
    
    // 根据负载均衡策略选择工作者
    switch wp.config.LoadBalancing {
    case "least_loaded":
        selectedWorker = wp.selectLeastLoadedWorker(workers)
    case "round_robin":
        selectedWorker = wp.selectRoundRobinWorker(workers)
    default:
        selectedWorker = wp.selectRandomWorker(workers)
    }
    
    if selectedWorker != nil {
        select {
        case selectedWorker.taskQueue <- job:
        case <-time.After(wp.config.WorkerTimeout):
            log.Printf("Failed to dispatch job %s: worker timeout", job.ID)
        case <-wp.ctx.Done():
            return
        }
    }
}

// selectLeastLoadedWorker 选择负载最小的工作者
func (wp *WorkerPool) selectLeastLoadedWorker(workers []*Worker) *Worker {
    var selected *Worker
    minLoad := int64(math.MaxInt64)
    
    for _, worker := range workers {
        load := atomic.LoadInt64(&worker.tasksProcessed)
        if load < minLoad {
            minLoad = load
            selected = worker
        }
    }
    
    return selected
}

// selectRoundRobinWorker 轮询选择工作者
func (wp *WorkerPool) selectRoundRobinWorker(workers []*Worker) *Worker {
    if len(workers) == 0 {
        return nil
    }
    
    // 简单的轮询实现
    index := int(atomic.AddInt64(&wp.metrics.TasksProcessed, 0)) % len(workers)
    return workers[index]
}

// selectRandomWorker 随机选择工作者
func (wp *WorkerPool) selectRandomWorker(workers []*Worker) *Worker {
    if len(workers) == 0 {
        return nil
    }
    
    index := rand.Intn(len(workers))
    return workers[index]
}

// Worker的start方法
func (w *Worker) start() {
    defer w.pool.workerWG.Done()
    
    log.Printf("Worker %d started", w.id)
    defer log.Printf("Worker %d stopped", w.id)
    
    for {
        select {
        case job := <-w.taskQueue:
            w.processJob(job)
            
        case <-time.After(w.pool.config.IdleTimeout):
            // 空闲超时，可能被移除
            continue
            
        case <-w.quit:
            return
            
        case <-w.pool.ctx.Done():
            return
        }
    }
}

// processJob 处理任务
func (w *Worker) processJob(job Job) {
    atomic.StoreInt32(&w.busy, 1)
    defer atomic.StoreInt32(&w.busy, 0)
    
    atomic.AddInt32(&w.pool.metrics.ActiveWorkers, 1)
    atomic.AddInt32(&w.pool.metrics.IdleWorkers, -1)
    
    defer func() {
        atomic.AddInt32(&w.pool.metrics.ActiveWorkers, -1)
        atomic.AddInt32(&w.pool.metrics.IdleWorkers, 1)
        
        atomic.AddInt64(&w.tasksProcessed, 1)
        w.lastActive = time.Now()
    }()
    
    startTime := time.Now()
    
    // 执行任务
    var result interface{}
    var err error
    
    if job.Timeout > 0 {
        // 带超时的任务执行
        done := make(chan struct{})
        go func() {
            defer close(done)
            result, err = job.Task()
        }()
        
        select {
        case <-done:
            // 任务完成
        case <-time.After(job.Timeout):
            err = fmt.Errorf("job %s timeout", job.ID)
        }
    } else {
        // 普通任务执行
        result, err = job.Task()
    }
    
    duration := time.Since(startTime)
    
    // 更新统计
    if err != nil {
        atomic.AddInt64(&w.pool.metrics.TasksFailed, 1)
    } else {
        atomic.AddInt64(&w.pool.metrics.TasksCompleted, 1)
    }
    
    // 更新平均延迟
    w.pool.updateAverageLatency(duration)
    
    // 调用回调函数
    if job.Callback != nil {
        go job.Callback(result, err)
    }
}

// autoScale 自动缩放
func (wp *WorkerPool) autoScale() {
    defer wp.workerWG.Done()
    
    ticker := time.NewTicker(wp.config.ScaleInterval)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            wp.performAutoScale()
            
        case <-wp.ctx.Done():
            return
        }
    }
}

// performAutoScale 执行自动缩放
func (wp *WorkerPool) performAutoScale() {
    if wp.scaling {
        return
    }
    
    wp.scaling = true
    defer func() { wp.scaling = false }()
    
    queueLen := len(wp.taskQueue)
    activeWorkers := int(atomic.LoadInt32(&wp.metrics.ActiveWorkers))
    totalWorkers := len(wp.workers)
    
    // 扩容条件：队列长度 > 工作者数量 * 2
    if queueLen > totalWorkers*2 && totalWorkers < wp.config.MaxWorkers {
        wp.addWorker()
        return
    }
    
    // 缩容条件：活跃工作者 < 总工作者 / 2 且队列为空
    if activeWorkers < totalWorkers/2 && queueLen == 0 && totalWorkers > wp.config.MinWorkers {
        wp.removeWorker()
        return
    }
}

// monitor 监控
func (wp *WorkerPool) monitor() {
    defer wp.workerWG.Done()
    
    ticker := time.NewTicker(wp.config.MetricsInterval)
    defer ticker.Stop()
    
    lastProcessed := int64(0)
    lastTime := time.Now()
    
    for {
        select {
        case <-ticker.C:
            currentProcessed := atomic.LoadInt64(&wp.metrics.TasksProcessed)
            currentTime := time.Now()
            
            // 计算吞吐量
            duration := currentTime.Sub(lastTime).Seconds()
            if duration > 0 {
                throughput := float64(currentProcessed-lastProcessed) / duration
                atomic.StoreUint64((*uint64)(&wp.metrics.ThroughputPerSec), 
                    math.Float64bits(throughput))
            }
            
            lastProcessed = currentProcessed
            lastTime = currentTime
            
            wp.logMetrics()
            
        case <-wp.ctx.Done():
            return
        }
    }
}

// updateAverageLatency 更新平均延迟
func (wp *WorkerPool) updateAverageLatency(latency time.Duration) {
    currentAvg := atomic.LoadInt64((*int64)(&wp.metrics.AverageLatency))
    newAvg := (time.Duration(currentAvg)*9 + latency) / 10
    atomic.StoreInt64((*int64)(&wp.metrics.AverageLatency), int64(newAvg))
}

// logMetrics 记录指标
func (wp *WorkerPool) logMetrics() {
    metrics := wp.GetMetrics()
    
    log.Printf("WorkerPool Metrics: "+
        "Workers=%d/%d, Queue=%d, Processed=%d, Completed=%d, Failed=%d, "+
        "AvgLatency=%v, Throughput=%.2f/s",
        metrics.ActiveWorkers+metrics.IdleWorkers, wp.config.MaxWorkers,
        metrics.QueueLength, metrics.TasksProcessed, metrics.TasksCompleted,
        metrics.TasksFailed, metrics.AverageLatency, metrics.ThroughputPerSec)
}

// GetMetrics 获取指标
func (wp *WorkerPool) GetMetrics() WorkerPoolMetrics {
    return WorkerPoolMetrics{
        ActiveWorkers:    atomic.LoadInt32(&wp.metrics.ActiveWorkers),
        IdleWorkers:      atomic.LoadInt32(&wp.metrics.IdleWorkers),
        QueueLength:      int64(len(wp.taskQueue)),
        TasksProcessed:   atomic.LoadInt64(&wp.metrics.TasksProcessed),
        TasksCompleted:   atomic.LoadInt64(&wp.metrics.TasksCompleted),
        TasksFailed:      atomic.LoadInt64(&wp.metrics.TasksFailed),
        AverageLatency:   time.Duration(atomic.LoadInt64((*int64)(&wp.metrics.AverageLatency))),
        ThroughputPerSec: math.Float64frombits(atomic.LoadUint64((*uint64)(&wp.metrics.ThroughputPerSec))),
    }
}

// Stop 停止工作池
func (wp *WorkerPool) Stop() error {
    log.Println("Stopping WorkerPool...")
    
    // 停止接受新任务
    wp.cancel()
    
    // 关闭任务队列
    close(wp.taskQueue)
    
    // 停止所有工作者
    wp.scalingMu.Lock()
    for _, worker := range wp.workers {
        close(worker.quit)
    }
    wp.scalingMu.Unlock()
    
    // 等待所有goroutine完成
    done := make(chan struct{})
    go func() {
        wp.workerWG.Wait()
        close(done)
    }()
    
    select {
    case <-done:
        log.Println("WorkerPool stopped gracefully")
        return nil
        
    case <-time.After(10 * time.Second):
        return fmt.Errorf("timeout waiting for workers to stop")
    }
}