go - 如何关闭 channel

Question

我试着改编这个例子: https://gobyexample.com/worker-pools

但我不知道如何停止 channel ，因为程序不会在 channel 循环结束时退出。

你能解释一下如何退出程序吗？

package main

import (
    "github.com/SlyMarbo/rss"
    "bufio"
    "fmt"
    "log"
    "os"
)

func readLines(path string) ([]string, error) {
    file, err := os.Open(path)
    if err != nil {
        return nil, err
    }
    defer file.Close()

    var lines []string
    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        lines = append(lines, scanner.Text())
    }
    return lines, scanner.Err()
}


func worker(id int, jobs <-chan string, results chan<- string) {
    for url := range jobs {
        fmt.Println("worker", id, "processing job", url)
        feed, err := rss.Fetch(url)
        if err != nil {
            fmt.Println("Error on: ", url)
            continue
        }
        borne := 0
        for _, value := range feed.Items {
            if borne < 5 {
                results <- value.Link
                borne = borne +1
            } else {
                continue
            }
        }
    }
}


func main() {
    jobs := make(chan string)
    results := make(chan string)

    for w := 1; w <= 16; w++ {
        go worker(w, jobs, results)
    }


    urls, err := readLines("flux.txt")
    if err != nil { 
        log.Fatalf("readLines: %s", err) 
    }

    for _, url := range urls {
        jobs <- url
    }

    close(jobs)

    // it seems program runs over...
    for msg := range results {
        fmt.Println(msg)
    }
}

flux.txt 是一个平面文本文件，如:

• http://blog.case.edu/news/feed.atom
• ...

Answer 1

问题是，在您提到的示例中，工作池从 results 中读取了 9 次:

for a := 1; a <= 9; a++ {
    <-results
}

另一方面，您的程序对 results 进行范围循环，这在 go 中具有不同的语义。范围运算符在 channel 关闭之前不会停止。

for msg := range results {
    fmt.Println(msg)
}

要解决您的问题，您需要关闭 results channel 。但是，如果您只是在 for 循环之前调用 close(results)，您很可能会 panic ，因为工作人员可能正在写 results。

要解决此问题，您需要添加另一个 channel ，以便在所有工作人员完成时收到通知。您可以使用 sync.WaitGroup 或:

const (
    workers = 16
)

func main() {
    jobs := make(chan string, 100)
    results := make(chan string, 100)
    var wg sync.WaitGroup

    for w := 0; w < workers; w++ {
        go func() {
            wg.Add(1)
            defer wg.Done()
            worker(w, jobs, results)
        }()
    }

    urls, err := readLines("flux.txt")
    if err != nil {
        log.Fatalf("readLines: %s", err)
    }

    for _, url := range urls {
        jobs <- url
    }

    close(jobs)

    wg.Wait()

    close(results)

    // it seems program runs over...
    for msg := range results {
        fmt.Println(msg)
    }
}

或者一个done channel :

package main

import (
    "bufio"
    "fmt"
    "github.com/SlyMarbo/rss"
    "log"
    "os"
)

func readLines(path string) ([]string, error) {
    file, err := os.Open(path)
    if err != nil {
        return nil, err
    }
    defer file.Close()

    var lines []string
    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        lines = append(lines, scanner.Text())
    }
    return lines, scanner.Err()
}

func worker(id int, jobs <-chan string, results chan<- string, done chan struct{}) {
    for url := range jobs {
        fmt.Println("worker", id, "processing job", url)
        feed, err := rss.Fetch(url)
        if err != nil {
            fmt.Println("Error on: ", url)
            continue
        }
        borne := 0
        for _, value := range feed.Items {
            if borne < 5 {
                results <- value.Link
                borne = borne + 1
            } else {
                continue
            }
        }
    }
    close(done)
}

const (
    workers = 16
)

func main() {
    jobs := make(chan string, 100)
    results := make(chan string, 100)
    dones := make([]chan struct{}, workers)

    for w := 0; w < workers; w++ {
        dones[w] = make(chan struct{})
        go worker(w, jobs, results, dones[w])
    }

    urls, err := readLines("flux.txt")
    if err != nil {
        log.Fatalf("readLines: %s", err)
    }

    for _, url := range urls {
        jobs <- url
    }


    close(jobs)

    for _, done := range dones {
        <-done
    }

    close(results)

    // it seems program runs over...
    for msg := range results {
        fmt.Println(msg)
    }
}