4. Go 并发编程基础：goroutine、channel、sync.WaitGroup、select 多路复用

在构建高性能 Kafka 压测工具时，并发是核心能力。Go 语言通过 goroutine 和 channel 提供了轻量级的并发模型，配合 sync.WaitGroup 和 select 多路复用机制，能够高效地编排压测任务。本章将系统讲解这四大并发原语，为后续实现高吞吐消息生产与消费奠定基础。

4.1 goroutine：轻量级并发执行单元

goroutine 是 Go 运行时管理的轻量级线程，创建成本极低（初始栈仅几 KB），可轻松启动数十万个并发任务。与操作系统线程不同，goroutine 由 Go 调度器在用户态进行多路复用，切换开销远小于线程。

// 启动一个匿名函数的 goroutine
go func() {
    fmt.Println("Hello from goroutine")
}()

// 启动命名函数的 goroutine
func sendMessage(topic string) {
    // 模拟发送消息到 Kafka
    time.Sleep(100 * time.Millisecond)
    fmt.Printf("Sent message to topic: %s\n", topic)
}

go sendMessage("test-topic")

4.2 channel：goroutine 之间的通信管道

Go 的并发哲学是“不要通过共享内存来通信，而应通过通信来共享内存”。channel 正是实现这一理念的核心类型，它提供了 goroutine 之间安全传递数据的管道。

// 无缓冲 channel 示例：同步传递消息
msgChan := make(chan string)

go func() {
    msgChan <- "Kafka message payload"  // 发送，阻塞直到接收方就绪
}()

received := <-msgChan  // 接收，阻塞直到发送方就绪
fmt.Println("Received:", received)

// 有缓冲 channel 示例：异步批量传递
batchChan := make(chan []byte, 100)  // 缓冲区容量 100

go func() {
    for i := 0; i < 10; i++ {
        batchChan <- []byte("record-" + strconv.Itoa(i))
    }
    close(batchChan)  // 发送完毕，关闭 channel
}()

for record := range batchChan {
    fmt.Printf("Processing record: %s\n", record)
}

4.3 sync.WaitGroup：等待 goroutine 完成

在压测场景中，主 goroutine 需要等待所有并发任务完成后才能汇总结果。sync.WaitGroup 提供了简洁的计数等待机制，用于协调一组 goroutine 的退出。

// Kafka 压测中并发发送消息的典型模式
func runConcurrentProducers(numProducers int, topic string) {
    var wg sync.WaitGroup

    for i := 0; i < numProducers; i++ {
        wg.Add(1)  // 每个生产者启动前增加计数
        go func(id int) {
            defer wg.Done()  // 确保 goroutine 退出时减少计数
            producer := createKafkaProducer()
            for j := 0; j < 1000; j++ {
                message := buildMessage(id, j)
                producer.Send(topic, message)
            }
            producer.Close()
            fmt.Printf("Producer %d finished\n", id)
        }(i)
    }

    wg.Wait()  // 等待所有生产者完成
    fmt.Println("All producers completed, collecting metrics...")
}

4.4 select 多路复用：同时监听多个 channel

select 语句允许一个 goroutine 同时等待多个 channel 操作，当任意一个 channel 就绪时执行对应的分支。这在压测工具中非常有用，例如同时处理消息发送结果、超时控制和退出信号。

// 压测主循环：同时处理消息发送、超时和退出信号
func runPressureTest(ctx context.Context, duration time.Duration) {
    timeout := time.After(duration)
    resultChan := make(chan SendResult, 1000)
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case result := <-resultChan:
            // 处理单个消息发送结果
            recordMetric(result)

        case <-ticker.C:
            // 每秒输出实时压测指标
            printCurrentMetrics()

        case <-timeout:
            // 压测时间到，优雅退出
            fmt.Println("Pressure test duration reached, stopping...")
            return

        case <-ctx.Done():
            // 收到外部取消信号（如 Ctrl+C）
            fmt.Println("Cancelled by user, cleaning up...")
            return

        default:
            // 所有 channel 都未就绪时，短暂休眠避免忙等待
            time.Sleep(10 * time.Millisecond)
        }
    }
}

4.5 四大原语在 Kafka 压测中的协同模式

在实际压测工具中，goroutine、channel、sync.WaitGroup 和 select 通常组合使用，形成完整的并发控制闭环。下表总结了各自的职责与协作关系：

// 综合示例：Kafka 压测并发框架骨架
func runBenchmark(config *BenchmarkConfig) {
    var wg sync.WaitGroup
    resultChan := make(chan ProducerResult, config.ChannelBuffer)
    ctx, cancel := context.WithTimeout(context.Background(), config.Duration)
    defer cancel()

    // 启动生产者 goroutine
    for i := 0; i < config.NumProducers; i++ {
        wg.Add(1)
        go func(producerID int) {
            defer wg.Done()
            runSingleProducer(ctx, producerID, config, resultChan)
        }(i)
    }

    // 启动结果收集 goroutine
    go func() {
        wg.Wait()
        close(resultChan)  // 所有生产者完成后关闭 channel
    }()

    // 主循环：select 多路复用处理结果和退出信号
    for {
        select {
        case result, ok := <-resultChan:
            if !ok {
                // channel 已关闭，所有生产者完成
                fmt.Println("All results collected, generating report...")
                return
            }
            aggregateResult(result)

        case <-ctx.Done():
            // 超时或取消
            fmt.Println("Benchmark finished or cancelled")
            return
        }
    }
}

4.6 常见并发模式与避坑指南