1.并发控制基础

Go语言中如何控制并发数量？

在Go语言中控制并发数主要有两种方式：使用带缓冲的channel和实现协程池。带缓冲的channel利用其阻塞特性来控制并发数量，当channel满时，新的goroutine会被阻塞。而协程池则是通过预先创建固定数量的worker goroutine来处理任务，这种方式可以更好地控制资源使用。

控制并发数就像是在管理一个餐厅的座位。如果餐厅有100个座位，但来了1000个客人，餐厅就会爆满，服务也会变差。同样，如果系统资源有限，但创建了太多goroutine，系统就会因为资源不足而变慢甚至崩溃。

想象一下，如果一个网站每秒要处理3万个请求，每个请求都创建一个goroutine，就像同时开3万个线程一样，电脑的内存很快就会用完，就像餐厅被挤爆一样。所以我们需要控制"同时工作的goroutine数量"，就像餐厅控制同时用餐的人数。

带缓冲的channel就像是一个有固定座位的等候区。当座位满了，新来的客人就要在外面等待。协程池就像是预先雇佣固定数量的服务员，不管有多少客人，都只有这些服务员在工作，这样既能保证服务质量，又不会让餐厅超负荷。

控制并发数的关键是找到平衡点：既要充分利用系统资源，又不能让系统过载。就像餐厅既要多接待客人赚钱，又要保证服务质量一样。

在实际应用中，对于简单的并发控制，使用带缓冲的channel就足够了。而对于复杂的业务场景，建议使用协程池，可以更好地控制资源使用。无论采用哪种方式，都要注意监控goroutine的数量，避免资源泄露。

控制并发数不是目的，而是手段。我们的最终目标是提高系统的整体性能和稳定性。因此，在选择控制方式时，要充分考虑业务特点和系统资源情况。

如何设计一个高效的goroutine池？

实现一个优雅的goroutine池需要考虑以下几个核心组件：任务队列、worker管理、任务调度和优雅退出机制。通过合理设计这些组件，我们可以实现一个高效且易用的goroutine池。

任务队列用 channel 实现，worker 通过循环从队列取任务。可以根据实际负载动态调整 worker 数量，避免资源紧张或浪费。优雅退出时，需确保所有任务处理完毕再关闭池子，防止任务丢失。

下面是一个简洁的实现示例：

type Pool struct {
    workers int
    tasks   chan Task
    wg      sync.WaitGroup
    stop    chan struct{}
}

func NewPool(workers int) *Pool {
    return &Pool{
        workers: workers,
        tasks:   make(chan Task, 1000),
        stop:    make(chan struct{}),
    }
}

func (p *Pool) Start() {
    for i := 0; i < p.workers; i++ {
        p.wg.Add(1)
        go p.worker()
    }
}

func (p *Pool) worker() {
    defer p.wg.Done()
    for {
        select {
        case task := <-p.tasks:
            task.Execute()
        case <-p.stop:
            return
        }
    }
}

在实际应用中，我们需要根据实际负载动态调整worker数量，实现任务优先级机制，添加监控指标，实现优雅退出机制。goroutine池的设计要遵循"简单但不过分简单"的原则。既要保证代码的可维护性，又要满足性能需求。

如何实现自适应的goroutine池？

动态调整的goroutine池能够根据系统负载和任务队列长度自动调整worker数量，实现资源的自适应管理。这种设计既避免了资源浪费，又保证了系统的响应能力。

核心设计思路是通过监控任务队列长度、处理时间和系统资源使用情况，动态调整worker数量。当任务积压严重时增加worker，当系统空闲时减少worker，实现负载均衡和资源优化。

监控指标包括任务队列长度、任务处理时间、系统CPU使用率、内存使用情况等。这些指标帮助池管理器做出正确的扩缩容决策。扩缩容策略应该平滑，避免频繁的调整导致系统不稳定。

核心实现示例：

type DynamicPool struct {
    minWorkers, maxWorkers, currentWorkers int
    tasks chan Task
    stop  chan struct{}
    wg    sync.WaitGroup
    mu    sync.RWMutex
}

func (p *DynamicPool) monitor() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            p.adjustWorkers()
        case <-p.stop:
            return
        }
    }
}

func (p *DynamicPool) adjustWorkers() {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    queueLen := len(p.tasks)
    current := p.currentWorkers
    
    // 扩容：队列长度超过阈值
    if queueLen > 100 && current < p.maxWorkers {
        p.scaleUp(2)
    }
    
    // 缩容：队列长度很低
    if queueLen < 10 && current > p.minWorkers {
        p.scaleDown(1)
    }
}

如何实现任务优先级调度？

带优先级的任务调度能够根据任务的紧急程度和重要性进行排序处理，确保关键任务优先执行。这种机制在实时系统、任务队列等场景中非常重要。

优先级队列设计使用堆数据结构实现，支持O(log n)的插入和删除操作。任务按优先级排序，高优先级任务优先执行。优先级可以是数值（数字越大优先级越高）或枚举类型（如紧急、高、中、低）。

任务饥饿问题是优先级调度中的常见问题，低优先级任务可能永远得不到执行。解决方案包括时间片轮转、优先级衰减、公平调度等策略。

// 优先级定义
type Priority int

const (
	PriorityLow Priority = iota    // 低优先级
	PriorityNormal                 // 正常优先级
	PriorityHigh                   // 高优先级
	PriorityUrgent                 // 紧急优先级
)

// 带优先级的任务结构
type Task struct {
	ID       int               // 任务ID
	Priority Priority          // 任务优先级
	Execute  func()            // 任务执行函数
}

// 优先级队列（实现堆接口）
type PriorityQueue []*Task

// 队列长度
func (pq PriorityQueue) Len() int           { return len(pq) }

// 优先级比较（数值大的优先级高）
func (pq PriorityQueue) Less(i, j int) bool { return pq[i].Priority > pq[j].Priority }

// 交换元素
func (pq PriorityQueue) Swap(i, j int)      { pq[i], pq[j] = pq[j], pq[i] }

// 添加任务到队列
func (pq *PriorityQueue) Push(x interface{}) {
	*pq = append(*pq, x.(*Task))
}

// 取出优先级最高的任务
func (pq *PriorityQueue) Pop() interface{} {
	old := *pq
	n := len(old)
	item := old[n-1]
	*pq = old[0 : n-1]
	return item
}

// 优先级调度器
type Scheduler struct {
	pq    PriorityQueue   // 优先级队列
	mu    sync.Mutex      // 互斥锁，保证队列操作安全
	stop  chan struct{}   // 停止信号
	wg    sync.WaitGroup  // 等待组，用于优雅退出
}

// 创建调度器（指定工作协程数量）
func NewScheduler(workers int) *Scheduler {
	s := &Scheduler{
		stop: make(chan struct{}),
	}
	// 初始化堆
	heap.Init(&s.pq)
	
	// 启动工作协程
	for i := 0; i < workers; i++ {
		s.wg.Add(1)
		go s.worker()
	}
	return s
}

// 提交任务到调度器
func (s *Scheduler) Submit(task *Task) {
	s.mu.Lock()
	heap.Push(&s.pq, task)  // 入队（自动按优先级排序）
	s.mu.Unlock()
}

// 工作协程逻辑
func (s *Scheduler) worker() {
	defer s.wg.Done()
	for {
		select {
		case <-s.stop:  // 收到停止信号，退出
			return
		default:
			s.mu.Lock()
			if s.pq.Len() > 0 {
				// 取出优先级最高的任务
				task := heap.Pop(&s.pq).(*Task)
				s.mu.Unlock()
				task.Execute()  // 执行任务
			} else {
				s.mu.Unlock()
				// 无任务时短暂休眠，减少CPU占用
				time.Sleep(10 * time.Millisecond)
			}
		}
	}
}

// 关闭调度器
func (s *Scheduler) Close() {
	close(s.stop)  // 发送停止信号
	s.wg.Wait()    // 等待所有工作协程退出
}

如何实现任务超时和取消？

任务超时和取消机制是并发控制中的重要组成部分，能够防止任务无限期执行，提高系统的响应性和稳定性。

超时控制通过context包实现，可以为任务设置执行超时时间。当超时发生时，任务会被自动取消，释放相关资源。超时机制特别适用于网络请求、数据库查询等可能长时间阻塞的操作。

取消机制允许外部主动取消正在执行的任务，通过context的取消信号实现。取消机制需要任务内部支持，定期检查取消信号并优雅退出。

type TimeoutExecutor struct {
    defaultTimeout time.Duration
    workers        int
    taskChan       chan Task
    stop           chan struct{}
    wg             sync.WaitGroup
}

type Task struct {
    ID      int
    Execute func(context.Context) error
    Timeout time.Duration
}

func (te *TimeoutExecutor) executeWithTimeout(task Task) {
    ctx, cancel := context.WithTimeout(context.Background(), task.Timeout)
    defer cancel()
    
    done := make(chan error, 1)
    go func() {
        done <- task.Execute(ctx)
    }()
    
    select {
    case err := <-done:
        fmt.Printf("任务 %d 执行结果: %v\n", task.ID, err)
    case <-ctx.Done():
        fmt.Printf("任务 %d 执行超时\n", task.ID)
    }
}

func cancellableTask(ctx context.Context) error {
    for i := 0; i < 100; i++ {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            time.Sleep(100 * time.Millisecond)
        }
    }
    return nil
}

如何实现任务重试和错误恢复？

任务重试和错误恢复机制能够提高系统的健壮性，处理临时性错误和网络抖动等问题。合理的重试策略能够显著提高任务的成功率。

重试策略包括固定间隔重试、指数退避重试、随机退避重试等。不同的错误类型需要不同的重试策略，临时性错误适合重试，永久性错误不应该重试。

错误分类是重试机制的基础，需要区分可重试错误和不可重试错误。网络超时、临时服务不可用等属于可重试错误，而参数错误、权限错误等属于不可重试错误。

type RetryStrategy struct {
    MaxAttempts   int
    InitialDelay  time.Duration
    MaxDelay      time.Duration
    BackoffFactor float64
}

type RetryableTask struct {
    ID       int
    Execute  func() error
    Strategy RetryStrategy
    OnRetry  func(attempt int, err error)
    OnFailure func(error)
}

type RetryExecutor struct {
    workers  int
    taskChan chan *RetryableTask
    stop     chan struct{}
    wg       sync.WaitGroup
}

func (re *RetryExecutor) executeWithRetry(task *RetryableTask) {
    var lastErr error
    
    for attempt := 1; attempt <= task.Strategy.MaxAttempts; attempt++ {
        if err := task.Execute(); err == nil {
            return
        } else {
            lastErr = err
            
            if !isRetryableError(err) {
                break
            }
            
            if attempt == task.Strategy.MaxAttempts {
                break
            }
            
            if task.OnRetry != nil {
                task.OnRetry(attempt, err)
            }
            
            delay := re.calculateDelay(task.Strategy, attempt)
            time.Sleep(delay)
        }
    }
    
    if task.OnFailure != nil {
        task.OnFailure(lastErr)
    }
}

func (re *RetryExecutor) calculateDelay(strategy RetryStrategy, attempt int) time.Duration {
    delay := float64(strategy.InitialDelay) * math.Pow(strategy.BackoffFactor, float64(attempt-1))
    
    if delay > float64(strategy.MaxDelay) {
        delay = float64(strategy.MaxDelay)
    }
    
    return time.Duration(delay)
}

func isRetryableError(err error) bool {
    if netErr, ok := err.(net.Error); ok {
        return netErr.Temporary() || netErr.Timeout()
    }
    return false
}