go - 如何从 go routine 中退出外循环？

Question

想法是从 go routine 中退出 outerloop，我使用了一个 channel 来发出中断循环的信号。我正在使用信号量模式来限制生成的 goroutine 数量，这样在等待循环退出时我不会生成大量的 go 例程。

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "sync"
)

type Task struct {
    ID        int    `json:"id"`
    UserID    int    `json:"user_id"`
    Title     string `json:"title"`
    Completed bool   `json:"completed"`
}

func main() {
    var t Task
    wg := &sync.WaitGroup{}
    stop := make(chan struct{})
    sem := make(chan struct{}, 10)

    results := make(chan Task, 1)

    worker := func(i int) {
        defer wg.Done()
        defer func() { <-sem }()
        res, err := http.Get(fmt.Sprintf("https://jsonplaceholder.typicode.com/todos/%d", i))
        if err != nil {
            log.Fatal(err)
        }
        defer res.Body.Close()
        if err := json.NewDecoder(res.Body).Decode(&t); err != nil {
            log.Fatal(err)
        }

        if i == 20 {
            close(stop)
        }
        results <- t
    }

    i := 0

outer:
    for {
        select {
        case <-stop:
            fmt.Println("I came here")
            close(sem)
            break outer
        case v := <-results:
            fmt.Println(v)
        default:
            wg.Add(1)
            sem <- struct{}{}
            go worker(i)
            i++
        }
    }
    wg.Wait()

    fmt.Println("I am done")
}

现在的问题是，我看到它进入了我试图打破循环的情况，但是它永远不会到达我完成了，原因可能是它在尝试时被无限阻塞收到结果。我想知道如何有效地处理同样的问题。

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "sync"
)

type Task struct {
    ID        int    `json:"id"`
    UserID    int    `json:"user_id"`
    Title     string `json:"title"`
    Completed bool   `json:"completed"`
}

func main() {
    wg := &sync.WaitGroup{}

    sem := make(chan struct{}, 10)
    ctx, cancel := context.WithCancel(context.Background())
    var ts []Task
    //results := make(chan Task, 1)

    worker := func(i int) {
        var t Task
        defer wg.Done()
        defer func() {
            <-sem
        }()
        res, err := http.Get(fmt.Sprintf("https://jsonplaceholder.typicode.com/todos/%d", i))
        if err != nil {
            log.Fatal(err)
        }
        defer res.Body.Close()
        if err := json.NewDecoder(res.Body).Decode(&t); err != nil {
            log.Fatal(err)
        }

        if i > 20 {
            cancel()
        }
        ts = append(ts, t)
    }

    i := 0

outer:
    for {
        select {
        case <-ctx.Done():
            break outer
        default:
            wg.Add(1)
            sem <- struct{}{}
            go worker(i)
            i++
        }
    }
    wg.Wait()

    fmt.Println(ts)
}

这行得通，但我最终在数组中得到了我想避免的重复条目。

编辑::@Davud 解决方案有效，但我仍然有兴趣知道进一步优化和限制生成的 goroutines 的数量。当前生成的额外 goroutines = sem 的缓冲区大小。我想在保持并发的同时减少它。

Answer 1

发生这种情况是因为一旦接收到停止信号并退出 for 循环，您就不再监听和打印结果，这会导致结果 channel 阻塞 worker 继续处理。

作为解决方案，您可以在单独的 goroutine 中收听结果 channel 。

我在这里删除了 case v := <-results: fmt.Println(v)并添加了一个 goroutine。试试看

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "sync"
)

type Task struct {
    ID        int    `json:"id"`
    UserID    int    `json:"user_id"`
    Title     string `json:"title"`
    Completed bool   `json:"completed"`
}

func main() {
    var t Task
    wg := &sync.WaitGroup{}
    stop := make(chan struct{})
    sem := make(chan struct{}, 10)

    results := make(chan Task, 1)

    worker := func(i int) {
        defer wg.Done()
        defer func() { <-sem }()
        res, err := http.Get(fmt.Sprintf("https://jsonplaceholder.typicode.com/todos/%d", i))
        if err != nil {
            log.Fatal(err)
        }
        defer res.Body.Close()
        if err := json.NewDecoder(res.Body).Decode(&t); err != nil {
            log.Fatal(err)
        }

        if i == 20 {
            close(stop)
        }
        results <- t
    }

    i := 0

    go func() {
        for v := range results {
            fmt.Println(v)
        }
    }()
outer:
    for {
        select {
        case <-stop:
            fmt.Println("I came here")
            close(sem)
            break outer
        default:
            wg.Add(1)
            sem <- struct{}{}
            go worker(i)
            i++
        }
    }
    wg.Wait()

    fmt.Println("I am done")
}