go - 无法获取工作 channel 图

Question

这可能是菜鸟的错误。我有一个带字符串值的 slice 和一个 channel 映射。对于 slice 中的每个字符串，将使用该字符串作为键创建一个 channel 并为其创建一个映射条目。

我观察 channel 并将值传递给其中一个，但从未发现过。

package main

import (
    "fmt"
    "time"
)

type TestStruct struct {
    Test string
}

var channelsMap map[string](chan *TestStruct)

func main() {
    stringsSlice := []string{"value1"}
    channelsMap := make(map[string](chan *TestStruct))

    for _, value := range stringsSlice {
        channelsMap[value] = make(chan *TestStruct, 1)

        go watchChannel(value)
    }

    <-time.After(3 * time.Second)

    testStruct := new(TestStruct)
    testStruct.Test = "Hello!"
    channelsMap["value1"] <- testStruct

    <-time.After(3 * time.Second)
    fmt.Println("Program ended")
}

func watchChannel(channelMapKey string) {
    fmt.Println("Watching channel: " + channelMapKey)

    for channelValue := range channelsMap[channelMapKey] {
        fmt.Printf("Channel '%s' used. Passed value: '%s'\n", channelMapKey, channelValue.Test)
    }

}

游乐场链接:https://play.golang.org/p/IbucTqMjdGO

输出:

Watching channel: value1
Program ended

当消息输入到 channel 中时，我该如何执行某些操作？

Answer 1

您的方法有很多问题。

第一个是您要重新声明(“屏蔽”)全局
您的channelsMap函数中的变量main。
(您是否至少完成了一些
most basic intro to Go，您应该没有这样的问题。)

这意味着您的watchChannel(实际上是执行该函数的所有goroutine)都会读取全局channelsMap，而main函数则将其写入本地channelsMap。

接下来发生的情况如下:

The range statement
在watchChannel中有一个简单的
映射查找表达式作为其来源channelsMap[channelMapKey]。

在Go中，这种形式的map lookup
永远不会失败，但是如果 map 没有这样的键(或者如果 map 未初始化，即为nil)，则所谓的
"zero value"
返回适当类型的。

由于全局channelsMap始终为空，因此对watchChannel的任何调用都会执行映射查找，该查找始终返回
类型为chan *TestStruct的零值。
任何 channel 的零值为nil。

通过range channel 执行的nil语句
produces zero iterations。
换句话说，for中的watchChannel循环总是执行
零次。

更复杂的问题仍然不是全局变量的影子，而是goroutine之间完全没有同步。您正在使用“ sleep ”作为一种创可贴，试图在goroutines之间执行隐式同步
但是虽然这似乎可以通过所谓的
“常识”，两个人在实践中都行不通
原因:

sleep 始终是幼稚的同步方法，因为它仅表明所有goroutine将相对自由地运行且不受竞争的事实。在许多(如果不是大多数)生产设置中，事实并非如此，因此，这始终是细微错误的原因。请不要再这样做了。

Go memory model中什么都没有
他说，等待壁钟时间是运行时认为建立不同goroutine的执行如何相互关联的顺序。

存在多种在goroutine之间同步执行的方法。基本上，它们等于使用sync包提供的类型通过 channel 发送和接收。
在您的特定情况下，最简单的方法可能是使用sync.WaitGroup类型。

Here是我们想要的
解决上述问题后，请执行以下操作:
-在 map 变量的位置对其进行初始化
定义，不要在main中弄乱它。
-使用sync.WaitGroup使其正确等待所有
它产生的goroutine信号完成了:

package main

import (
    "fmt"
    "sync"
)

type TestStruct struct {
    Test string
}

var channelsMap = make(map[string](chan *TestStruct))

func main() {
    stringsSlice := []string{"value1"}

    var wg sync.WaitGroup

    wg.Add(len(stringsSlice))
    for _, value := range stringsSlice {
        channelsMap[value] = make(chan *TestStruct, 1)

        go watchChannel(value, &wg)
    }

    testStruct := new(TestStruct)
    testStruct.Test = "Hello!"
    channelsMap["value1"] <- testStruct

    wg.Wait()
    fmt.Println("Program ended")
}

func watchChannel(channelMapKey string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Watching channel: " + channelMapKey)

    for channelValue := range channelsMap[channelMapKey] {
        fmt.Printf("Channel '%s' used. Passed value: '%s'\n", channelMapKey, channelValue.Test)
    }

}

一旦我们将您的代码接下来的两个问题变得显而易见
解决了前两个问题-在您制作了“观察者” goroutine之后
使用与运行main的goroutine相同的map变量，并且
让后者适当地等待观察者:

有一个data race
在 map 变量之间
在main循环产生后更新 map 的代码
观察者goroutine结束，访问此例程的代码
所有观察者goroutine中的变量。

有一个deadlock
在监视程序和等待它们完成的主程序之间。

僵局的原因是观察者goroutines
永远不会收到任何必须退出处理的信号，
因此永远被困在试图从各自的阅读中
channel 。

解决这两个新问题的方法很简单，但是它们
可能实际上会“破坏”您最初的结构构想
您的代码。

首先，我只是通过让观察者来消除数据竞争
无法访问 map 变量。如您所见，每次致电for接收单个值用作 key
从共享 map 中读取值，因此每个观察者始终
在其运行时仅读取一次单个值。
如果我们删除多余的代码，代码将变得更加清晰
完全访问 map ，而是通过相应的 channel
直接给每个观察者带来值(value)。
一个不错的副产品是我们不需要全局
映射变量了。

Here，我们将获得:

package main

import (
    "fmt"
    "sync"
)

type TestStruct struct {
    Test string
}

func main() {
    stringsSlice := []string{"value1"}
    channelsMap := make(map[string](chan *TestStruct))

    var wg sync.WaitGroup

    wg.Add(len(stringsSlice))
    for _, value := range stringsSlice {
        channelsMap[value] = make(chan *TestStruct, 1)

        go watchChannel(value, channelsMap[value], &wg)
    }

    testStruct := new(TestStruct)
    testStruct.Test = "Hello!"
    channelsMap["value1"] <- testStruct

    wg.Wait()
    fmt.Println("Program ended")
}

func watchChannel(channelMapKey string, ch <-chan *TestStruct, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Watching channel: " + channelMapKey)

    for channelValue := range ch {
        fmt.Printf("Channel '%s' used. Passed value: '%s'\n", channelMapKey, channelValue.Test)
    }

}

好吧，我们仍然有僵局。

有多种方法可以解决此问题，但它们取决于
根据实际情况，并以这个玩具示例为例，
尝试迭代至少其中一部分
弄乱水。
相反，在这种情况下，我们使用最简单的方法:关闭
channel 立即对其进行任何挂起的接收操作
解锁并产生 channel 类型的零值。
对于使用watchChannel语句进行迭代的 channel
它只是意味着结局终止而没有产生任何
channel 的值(value)。

换句话说，让我们关闭所有 channel 以解除阻止
观察者goroutines正在运行的range语句
然后等待这些goroutine通过wait组报告其完成情况。

为了不使答案太长，我还添加了字符串 slice 的程序初始化，通过让多个观察者(而不仅仅是一个观察者)实际做有用的工作，使示例更加有趣:

package main

import (
    "fmt"
    "sync"
)

type TestStruct struct {
    Test string
}

func main() {
    var stringsSlice []string
    channelsMap := make(map[string](chan *TestStruct))

    for i := 1; i <= 10; i++ {
        stringsSlice = append(stringsSlice, fmt.Sprintf("value%d", i))
    }

    var wg sync.WaitGroup

    wg.Add(len(stringsSlice))
    for _, value := range stringsSlice {
        channelsMap[value] = make(chan *TestStruct, 1)

        go watchChannel(value, channelsMap[value], &wg)
    }

    for _, value := range stringsSlice {
        testStruct := new(TestStruct)
        testStruct.Test = fmt.Sprint("Hello! ", value)
        channelsMap[value] <- testStruct
    }

    for _, ch := range channelsMap {
        close(ch)
    }

    wg.Wait()
    fmt.Println("Program ended")
}

func watchChannel(channelMapKey string, ch <-chan *TestStruct, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Watching channel: " + channelMapKey)

    for channelValue := range ch {
        fmt.Printf("Channel '%s' used. Passed value: '%s'\n", channelMapKey, channelValue.Test)
    }

}

Playground link。

如您所见，有些事情您应该实际学习
在着手与
并发。

我建议按以下顺序进行:

The Go tour会让您习惯于并发。

The Go Programming Language分为两章，专门为读者提供使用range包中的 channel 和类型来解决并发问题的简要介绍。

Concurrency In Go继续介绍了人们如何在Go中处理并发性的更多核心细节，包括解决并发程序在生产中面临的现实问题的高级主题，例如对传入请求进行速率限制的方法。