Go语言批量创建Goroutine处理十万条记录的Pub/Sub与DB更新问题

1. 现有代码无法完成所有记录的状态更新 #

你的代码存在严重逻辑错误：

• 外层循环每次仅遍历单条record，却尝试从pubChan接收len(leads)次数据，但每次只启动了1个goroutine发送1条数据到通道，这会导致程序永久阻塞在接收通道的循环中，根本无法推进到数据库更新步骤。
• 即使忽略阻塞问题，外层循环每次仅处理1条记录却执行一次批量更新，既浪费了批量更新的性能优势，也无法覆盖所有记录的状态更新逻辑。

2. 大量Goroutine的内存与性能风险 #

如果直接给10万条记录各启动一个goroutine，单个goroutine内存开销虽不大（约2KB），10万个总开销约200MB，一般不会直接内存溢出，但会引发两个核心问题：

• Go调度器需要频繁切换大量goroutine，显著降低整体执行效率。
• Pub/Sub API存在并发调用限制，大量并发Publish请求会触发限流，导致请求失败率上升，反而拖慢处理速度。

3. 优化实现方案：控制并发+批量处理 #

推荐使用固定大小的worker池控制并发数，同时批量收集结果后统一更新数据库，既避免goroutine泛滥，又能提升处理效率。

优化代码示例 #

// 定义并发worker数量，可根据Pub/Sub配额和服务器性能调整，比如设为50
const workerCount = 50

type pubRes struct {
    RecordId string
    Error    error
}

func processRecords(ctx context.Context, records []Record, topicName string) error {
    // 1. 创建任务通道和结果通道
    taskChan := make(chan Record, len(records))
    resultChan := make(chan pubRes, len(records))

    // 2. 启动固定数量的worker goroutine
    for w := 0; w < workerCount; w++ {
        go func() {
            for record := range taskChan {
                // 发送记录到Pub/Sub
                pubSubErr := pubsub.Publish(ctx, topicName, map[string]string{
                    "id": record.Id,
                })
                resultChan <- pubRes{
                    RecordId: record.Id,
                    Error:    pubSubErr,
                }
            }
        }()
    }

    // 3. 异步发送所有任务到通道，避免阻塞主线程
    go func() {
        for _, record := range records {
            taskChan <- record
        }
        close(taskChan) // 任务发送完毕，关闭通道让worker自动退出
    }()

    // 4. 收集所有处理结果
    var successRecordIds []string
    for i := 0; i < len(records); i++ {
        res := <-resultChan
        if res.Error == nil {
            successRecordIds = append(successRecordIds, res.RecordId)
        }
    }
    close(resultChan)

    // 5. 批量更新成功记录的状态
    if err := BulkUpdateStatus(ctx, successRecordIds, "success"); err != nil {
        return err
    }

    // 可选：单独收集失败记录ID，批量更新状态为"failed"
    return nil
}

额外优化方向 #

• Pub/Sub批量发送：如果使用的Pub/Sub服务支持批量消息（如Google Cloud Pub/Sub的批量Publish），可将多条记录打包成批量请求，减少API调用次数。
• 分批次处理：若10万条记录一次性处理内存压力大，可将记录拆分为多个批次（如每批1万条），逐批次执行上述流程。
• 上下文超时控制：在ctx中设置合理超时时间，避免任务因网络问题无限期阻塞。

内容的提问来源于stack exchange，提问作者aakash singh