Go语言中实现弹性工作池：从无限队列中持续处理任务

package main

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

// Work represents a single job to be processed.
type Work struct {
    ID string
}

// Worker processes jobs from workChan or signals readiness on requestChan.
// id: worker的唯一标识
// workChan: 从此通道接收任务
// requestChan: 当空闲时，向此通道发送信号
// wg: 用于主协程等待所有worker完成的WaitGroup
func Worker(id int, workChan <-chan Work, requestChan chan<- struct{}, wg *sync.WaitGroup) {
    defer wg.Done() // 确保在worker退出时通知WaitGroup
    fmt.Printf("Worker %d started.\n", id)

    for {
        select {
        case work, ok := <-workChan:
            // 尝试从workChan接收任务
            if !ok {
                // workChan已关闭，表示没有更多任务，worker可以安全退出
                fmt.Printf("Worker %d: workChan closed, exiting.\n", id)
                return
            }
            // 模拟任务处理
            fmt.Printf("Worker %d: Processing work %s\n", id, work.ID)
            time.Sleep(time.Millisecond * 200) // 模拟耗时操作

        case requestChan <- struct{}{}:
            // 当workChan没有任务时，此case会被选中
            // worker发送一个空结构体{}到requestChan，表示自己已空闲并请求更多任务
            // 使用 struct{}{} 是因为它不占用内存，只用于信号传递
            // 注意：为了避免输出过于频繁，这里通常不打印日志
            // fmt.Printf("Worker %d: Signaling idle for more work.\n", id)
        }
    }
}

// Dispatcher manages the work flow.
// numWorkers: 工作协程的数量
// batchSize: 每次从队列中获取的任务批次大小
// totalJobs: 模拟的“永不停止”队列中的总任务数（实际应用中可能真是无限）
func Dispatcher(numWorkers, batchSize int, totalJobs int) {
    // workChan: 带有缓冲的任务通道，用于存储待处理的任务批次
    workChan := make(chan Work, batchSize)
    // requestChan: 带有缓冲的请求通道，用于接收worker的空闲信号
    // 缓冲大小等于worker数量，确保每个worker都能发送其空闲信号
    requestChan := make(chan struct{}, numWorkers)

    var wg sync.WaitGroup

    // 启动所有工作协程
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go Worker(i, workChan, requestChan, &wg)
    }

    jobsFetched := 0     // 已经从模拟队列中获取的任务总数
    jobsProcessed := 0   // 调度器记录的已处理任务总数（通过requestChan信号粗略估计）

    // 调度器主循环，持续获取和分发任务
    for jobsProcessed < totalJobs {
        // 策略：当任务通道的缓冲接近一半空闲时，或者当所有任务尚未完全获取时，尝试获取新的任务批次。
        // 这样可以保持workChan总是有任务，减少worker等待时间。
        if len(workChan) < batchSize/2 && jobsFetched < totalJobs {
            // 计算本次需要获取的任务数量
            jobsToFetch := min(batchSize, totalJobs-jobsFetched)
            if jobsToFetch > 0 {
                fmt.Printf("\nDispatcher: Fetching %d new jobs...\n", jobsToFetch)
                for i := 0; i < jobsToFetch; i++ {
                    jobID