跨渠道交流

Question

这个程序简单地计算“某物”在单独的峡谷和主峡谷最终结束后，所有的峡谷都完成了。

我想问的问题是，在printChannels()中退出无限for循环的方式是可以接受的吗？还是有人有更好的方法来做这件事？

package concurrency

import (
    "goreceipes/concurrency/syncutils"
    "fmt"
    "math"
)

// create a struct to hold the number and it's processed value
type NumberObject struct {
    number int
    value  int
}

type QuitObject struct {
    channelName string
    quitValue int
}

// Main
func main() {
    // notify the main of 4 threads in play
    syncutils.Wg.Add(4)

    // create channels for each function
    squareCh := make(chan NumberObject)
    fibCh := make(chan NumberObject)
    dblCh := make(chan NumberObject)
    quitCh := make(chan QuitObject, 3)

    // launch threads to calculate values
    go calculateSquares(squareCh, quitCh)
    go calculateFibonacci(fibCh, quitCh)
    go calculateDouble(dblCh, quitCh)

    // launch the printer thread
    go printChannels(squareCh, fibCh, dblCh, quitCh)

    // wait for threads to complete
    syncutils.Wg.Wait()

    fmt.Println("Terminating program.")
}

// print the output of each channel
func printChannels(sqCh <-chan NumberObject, fibCh <-chan NumberObject, dblCh <-chan NumberObject, quitCh <- chan QuitObject) {
    // let the 'main' know i'm done
    defer syncutils.Wg.Done()

    // maintains a count of how many channels are exhausted
    channelMap := make(map[string]int)

    // initialize with zero. After each goroutine is done executing, they will send a "1" into the map as a "quit signal"
    channelMap["sqCh"] = 0
    channelMap["fibCh"] = 0
    channelMap["dblCh"] = 0

    for {
        select {
        case obj := <- sqCh:
            fmt.Printf("Square of %d = \t%d\n", obj.number, obj.value)
        case obj := <- fibCh:
            fmt.Printf("Fibonacci of %d = %d\n", obj.number, obj.value)
        case obj := <- dblCh:
            fmt.Printf("Double of %d = \t%d\n", obj.number, obj.value)
        case val := <- quitCh:
            channelMap[val.channelName] = val.quitValue
            if channelMap["sqCh"] == 1 && channelMap["fibCh"] == 1 && channelMap["dblCh"] == 1{
                fmt.Println("All channels are done executing. Break the infinite loop")
                return
            }
        }
    }
}

// calculates double
func calculateDouble(dblCh chan<- NumberObject, quitCh chan <- QuitObject) {
    defer syncutils.Wg.Done()

    for i := 0; i < 10; i++ {
        dblCh <- NumberObject{number: i, value: i * 2}
    }
    // send the quit signal
    quitCh <- QuitObject{"dblCh", 1}
}

// calculate fibonacci
func calculateFibonacci(fibCh chan<- NumberObject, quitCh chan <- QuitObject) {
    // let the main know I'm done
    defer syncutils.Wg.Done()

    for i := 0; i < 10; i++ {
        num := float64(i)
        Phi := (1 + math.Sqrt(num)) / 2
        phi := (1 - math.Sqrt(num)) / 2
        result := (math.Pow(Phi, num) - math.Pow(phi, num)) / math.Sqrt(5)
        fibCh <- NumberObject{number: int(num), value: int(result)}
    }
    // send the quit signal
    quitCh <- QuitObject{"fibCh", 1}
}

// calculates squares
func calculateSquares(sqCh chan<- NumberObject, quitCh chan <- QuitObject) {
    // let the main know I'm done
    defer syncutils.Wg.Done()

    for i := 0; i < 10; i++ {
        sqCh <- NumberObject{number: i, value: i * i}
    }
    // send the quit signal
    quitCh <- QuitObject{"sqCh", 1}
}

和

syncutils.Wg

在一个单独的包中定义，因此它可以在任何需要的地方使用。

package syncutils

import "sync"

var Wg sync.WaitGroup

/**
    the purpose of this file/code is to only provide a global variable Wg = WaitGroup and package it
    so it can be included wherever needed.
 */

有人能提出一个更好的方法来终止一个无限循环，即在一个不同的循环中读取多个通道吗？

提前谢谢你的帮助。

Answer 1

您可以不使用chan QuitObject：

正如您提到的，主要问题是正确退出printChannels。由于它是通过三个频道收听，所以它的实现相当复杂。

为了解决这个问题，我建议添加一个chan string，它将(同时)接收要从其他三个通道打印的字符串，并在它们完成后关闭。

printChannels变成printResult (实际上可以直接在main中编写)：

func printResult(in <-chan string) {
    for s := range in { // this loop will end once the input channel is closed
        fmt.Println(s)
    }
}

为了满足这个通道，我们需要goroutines获取结果并创建一个字符串：

func formatResult(in <-chan NumberObject, out chan<- string, format string) {
    for n := range in { // this loop will end once the input channel is closed
        out <- fmt.Sprintf(format, n.number, n.value)
    }
}
// usage:
resultCh := make(chan string)
go formatResult(squareCh, resultCh, "Square of %d = \t%d")
go formatResult(fibCh, resultCh, "Fibonacci of %d = %d")
go formatResult(dblCh, resultCh, "Double of %d = \t%d")
// close the resultCh at the end

实际上，我们刚刚转移了问题:我们现在需要知道什么时候关闭resultCh！

在这里，WaitGroup是有意义的(我们只需要一个局部变量)：

var wg = sync.WaitGroup{}
wg.Add(3)
go func() {
    formatResult(squareCh, resultCh, "Square of %d = \t%d")
    wg.Done()
}()
go func() {
    formatResult(fibCh, resultCh, "Fibonacci of %d = %d")
    wg.Done()
}()
go func() {
    formatResult(dblCh, resultCh, "Double of %d = \t%d")
    wg.Done()
}()

go func() {
    // wait for threads to complete
    wg.Wait()
    resultCh <- "All channels are done executing."
    close(resultCh)
}()

但实际上，formatResult从未完成，因为输入通道从未关闭过.因此，我们需要修改计算函数，以关闭通道后的for循环。

// calculates squares
func calculateSquares(sqCh chan<- NumberObject) {
    for i := 0; i < 10; i++ {
        sqCh <- NumberObject{number: i, value: i * i}
    }
    close(sqCh)
}

作为进一步的改进，您可以将计算函数更改为func(int)int，并添加辅助函数来调用它们并填充预期的通道：

func calculate(f func(int) int, ch chan<- NumberObject) {
    for i := 0; i < 10; i++ {
        ch <- NumberObject{number: i, value: f(i)}
    }
    close(ch)
}

最后一个程序(我将wg.Done集成在一个calculateAndFormat中以减少详细，但它稍微降低了一点灵活性)：

package main

import (
    "fmt"
    "math"
    "sync"
)

func main() {
    resultCh := make(chan string)

    var wg = sync.WaitGroup{}
    wg.Add(3)
    // launch threads to calculate values
    go calculateAndFormat(square, resultCh, "Square of %d = \t%d", &wg)
    go calculateAndFormat(fibonacci, resultCh, "Fibonacci of %d = %d", &wg)
    go calculateAndFormat(double, resultCh, "Double of %d = \t%d", &wg)

    go func() {
        wg.Wait()
        resultCh <- "All channels are done executing."
        close(resultCh)
    }()

    for s := range resultCh {
        fmt.Println(s)
    }

    fmt.Println("Terminating program.")
}

func calculateAndFormat(f func(int) int, ch chan<- string, format string, wg *sync.WaitGroup) {
    for i := 0; i < 10; i++ {
        ch <- fmt.Sprintf(format, i, f(i))
    }
    wg.Done()
}

// calculates double
func double(i int) int {
    return i * 2
}

// calculate fibonacci
func fibonacci(i int) int {
    num := float64(i)
    Phi := (1 + math.Sqrt(num)) / 2
    phi := (1 - math.Sqrt(num)) / 2
    result := (math.Pow(Phi, num) - math.Pow(phi, num)) / math.Sqrt(5)
    return int(result)
}

// calculates squares
func square(i int) int {
    return i * i
}