寻求更简洁的Go并发实现方案：多外部API调用场景

需求 #

我需要实现一个可并发调用多个外部API的函数，满足以下要求：

• 返回首个遇到的错误
• 聚合所有API调用的结果

非并发版本实现 #

func GetMicrosoftTeamsChannelSuggestions(workspace *model.Workspace) ([]*model.MicrosoftTeamsChannel, error) {
    allChannels := []*model.MicrosoftTeamsChannel{}
    teamsGroups := GetMicrosoftTeamsGroupsFromWorkspace(workspace)

    for _, teamGroup := range teamsGroups {
        channels, err := GetMicrosoftTeamsChannels(*workspace.MicrosoftTeamsTenantId, teamGroup)
        if err != nil {
            return nil, err
        }
        allChannels = append(allChannels, channels...)
    }

    return allChannels, nil
}

当前并发实现（逻辑繁琐） #

func GetMicrosoftTeamsChannelSuggestions(workspace *model.Workspace) ([]*model.MicrosoftTeamsChannel, error) {
    allChannels := []*model.MicrosoftTeamsChannel{}
    teamsGroups := GetMicrosoftTeamsGroupsFromWorkspace(workspace)

    ch := make(chan []*model.MicrosoftTeamsChannel, len(teamsGroups))
    errCh := make(chan error)

    defer func() {
        close(ch)
        close(errCh)
    }()

    var wg sync.WaitGroup
    wg.Add(len(teamsGroups))

    for _, teamGroup := range teamsGroups {
        go func(teamGroup string) {
            defer wg.Done()

            channels, err := GetMicrosoftTeamsChannels(*workspace.MicrosoftTeamsTenantId, teamGroup)
            if err != nil {
                errCh <- err
            } else {
                ch <- channels
            }
        }(teamGroup)
    }

    wg.Wait()

    for {
        select {
        case channels := <-ch:
            allChannels = append(allChannels, channels...)
        case err := <-errCh:
            return nil, err // Return the first encountered error
        default:
            return allChannels, nil
        }
    }
}

诉求 #

当前方案包含繁琐的循环和select多路复用逻辑，希望得到更简洁的实现方式。

简洁实现方案 #

可以通过单个带缓冲的结果通道统一传递结果和错误，结合context实现快速失败（一旦有错误就取消所有未完成的goroutine），避免无效API调用：

import (
    "context"
    "sync"
)

func GetMicrosoftTeamsChannelSuggestions(workspace *model.Workspace) ([]*model.MicrosoftTeamsChannel, error) {
    teamsGroups := GetMicrosoftTeamsGroupsFromWorkspace(workspace)
    if len(teamsGroups) == 0 {
        return []*model.MicrosoftTeamsChannel{}, nil
    }

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // 确保函数退出时取消上下文

    // 定义统一的结果结构体
    type result struct {
        channels []*model.MicrosoftTeamsChannel
        err      error
    }
    resCh := make(chan result, len(teamsGroups))

    var wg sync.WaitGroup
    wg.Add(len(teamsGroups))

    for _, teamGroup := range teamsGroups {
        go func(tg string) {
            defer wg.Done()

            // 先检查上下文是否已取消，避免无效调用
            select {
            case <-ctx.Done():
                return
            default:
                channels, err := GetMicrosoftTeamsChannels(*workspace.MicrosoftTeamsTenantId, tg)
                resCh <- result{channels: channels, err: err}
            }
        }(teamGroup)
    }

    // 后台等待所有goroutine完成后关闭通道
    go func() {
        wg.Wait()
        close(resCh)
    }()

    allChannels := []*model.MicrosoftTeamsChannel{}
    // 遍历通道接收结果，遇到错误立即返回
    for res := range resCh {
        if res.err != nil {
            cancel() // 取消剩余goroutine，终止未完成的API调用
            return nil, res.err
        }
        allChannels = append(allChannels, res.channels...)
    }

    return allChannels, nil
}

方案优势 #

• 用单个result通道统一处理结果和错误，无需维护多个独立通道
• 通过context.WithCancel实现快速失败，减少不必要的资源消耗
• 结果接收逻辑简化为普通通道遍历，去掉复杂的select+default组合
• 代码结构清晰，逻辑可读性更强

内容的提问来源于stack exchange，提问作者Jemilu Mohammed