Golang goroutine不会在通道内运行

huangapple
go评论118阅读模式
英文:

Golang goroutine doesn't run with channel inside

问题

我正在尝试实现一个单词计数程序，但在第一步遇到了一些问题。

这是我的代码：

  1. package main
  3. import (
  4.     "fmt"
  5.     "os"
  6.     "bufio"
  7.     "sync"
  8. )
  10. // 将数据加载到通道中
  11. func laodData(arr []string,channel chan string,wg sync.WaitGroup) {
  12.     for _,path := range arr {
  13.         file,err := os.Open(path)
  14.         fmt.Println("开始加载数据 ", path)
  15.         if err != nil {
  16.             fmt.Println(err)
  17.             os.Exit(-1)
  18.         }
  19.         defer file.Close()
  20.         reader := bufio.NewReaderSize(file, 32*10*1024)
  21.         i := 0
  22.         for {
  23.             line,err := reader.ReadString('\n')
  24.             channel <- line
  25.             if err != nil {
  26.                 break
  27.             }
  28.             i++
  29.             if i%200 == 0 {
  30.                 fmt.Println(i, "行已解析")
  31.             }
  32.         }
  33.         fmt.Println("完成数据加载 ", path)
  34.     }
  35.     wg.Done()
  36. }
  38. // 将数据行分发给不同的映射器
  39. func dispatcher(channel chan string,wg sync.WaitGroup){
  40.     fmt.Println("拉取数据 11")
  41.     line,ok := <- channel
  42.     fmt.Println(ok)
  43.     for ok {
  44.         fmt.Println(line)
  45.         line,ok = <- channel
  46.     }
  47.     fmt.Println("拉取数据 22")
  48.     wg.Done()
  49. }
  51. func main() {
  52.     path := os.Args
  53.     if len(path) < 2 {
  54.         fmt.Println("需要输入文件")
  55.         os.Exit(0)
  56.     }
  57.     var wg sync.WaitGroup
  58.     wg.Add(2)
  60.     channel := make(chan string)
  61.     defer close(channel)
  62.     
  63.     fmt.Println("在分发器之前")
  64.     go laodData(path[1:],channel,wg)
  65.     go dispatcher(channel,wg)
  66.     wg.Wait()
  68.     fmt.Println("在分发器之后")
  69. }

这是我的输出：

  1. ...
  3. 完成数据加载  result.txt
  5. 抛出异常：所有的goroutine都处于休眠状态 - 死锁！
  7. goroutine 1 [semacquire]:
  8. sync.runtime_Semacquire(0x42154100, 0x42154100)
  9.     /usr/local/go/src/pkg/runtime/zsema_amd64.c:146 +0x25
  10. sync.(*WaitGroup).Wait(0x4213b440, 0x0)
  11.     /usr/local/go/src/pkg/sync/waitgroup.go:79 +0xf2
  12. main.main()
  13.     /Users/kuankuan/go/src/mreasy/main.go:66 +0x238
  15. goroutine 2 [syscall]:
  16. created by runtime.main
  17.     /usr/local/go/src/pkg/runtime/proc.c:221
  19. goroutine 4 [chan receive]:
  20. main.dispatcher(0x42115a50, 0x0, 0x2, 0x0)
  21.     /Users/kuankuan/go/src/mreasy/main.go:45 +0x223
  22. created by main.main
  23.     /Users/kuankuan/go/src/mreasy/main.go:65 +0x228
  24. 退出状态 2

谢谢！

英文:

I'm trying to implement a word count program, but with the first step i got some problem.

Here's my code:

  1. package main
  3. import (
  4.     &quot;fmt&quot;
  5.     &quot;os&quot;
  6.     &quot;bufio&quot;
  7.     &quot;sync&quot;
  8. )
  10. // Load data into channel
  11. func laodData(arr []string,channel chan string,wg sync.WaitGroup) {
  12.     for _,path := range arr {
  13.         file,err := os.Open(path)
  14.         fmt.Println(&quot;begin to laodData &quot;, path)
  15.         if err != nil {
  16.             fmt.Println(err)
  17.             os.Exit(-1)
  18.         }
  19.         defer file.Close()
  20.         reader := bufio.NewReaderSize(file, 32*10*1024)
  21.         i := 0
  22.         for {
  23.             line,err := reader.ReadString(&#39;\n&#39;)
  24.             channel &lt;- line
  25.             if err != nil {
  26.                 break
  27.             }
  28.             i++
  29.             if i%200 == 0 {
  30.                 fmt.Println(i,&quot; lines parsed&quot;)
  31.             }
  32.         }
  33.         fmt.Println(&quot;finish laodData &quot;, path)
  34.     }
  35.     wg.Done()
  36. }
  38. // dispatch data lines into different mappers
  39. func dispatcher(channel chan string,wg sync.WaitGroup){
  40.     fmt.Println(&quot;pull data 11&quot;)
  41.     line,ok := &lt;- channel
  42.     fmt.Println(ok)
  43.     for ok {
  44.         fmt.Println(line)
  45.         line,ok = &lt;- channel
  46.     }
  47.     fmt.Println(&quot;pull data 22&quot;)
  48.     wg.Done()
  49. }
  51. func main() {
  52.     path := os.Args
  53.     if len(path) &lt; 2 {
  54.         fmt.Println(&quot;Need Input Files&quot;)
  55.         os.Exit(0)
  56.     }
  57.     var wg sync.WaitGroup
  58.     wg.Add(2)
  60.     channel := make(chan string)
  61.     defer close(channel)
  62.     
  63.     fmt.Println(&quot;before dispatcher&quot;)
  64.     go laodData(path[1:],channel,wg)
  65.     go dispatcher(channel,wg)
  66.     wg.Wait()
  68.     fmt.Println(&quot;after dispatcher&quot;)
  69. }

And here's my output:

  1. ...
  3. finish laodData  result.txt
  5. throw: all goroutines are asleep - deadlock!
  7. goroutine 1 [semacquire]:
  8. sync.runtime_Semacquire(0x42154100, 0x42154100)
  9. 	/usr/local/go/src/pkg/runtime/zsema_amd64.c:146 +0x25
  10. sync.(*WaitGroup).Wait(0x4213b440, 0x0)
  11. 	/usr/local/go/src/pkg/sync/waitgroup.go:79 +0xf2
  12. main.main()
  13. 	/Users/kuankuan/go/src/mreasy/main.go:66 +0x238
  15. goroutine 2 [syscall]:
  16. created by runtime.main
  17. 	/usr/local/go/src/pkg/runtime/proc.c:221
  19. goroutine 4 [chan receive]:
  20. main.dispatcher(0x42115a50, 0x0, 0x2, 0x0)
  21. 	/Users/kuankuan/go/src/mreasy/main.go:45 +0x223
  22. created by main.main
  23. 	/Users/kuankuan/go/src/mreasy/main.go:65 +0x228
  24. exit status 2

Thanks !

答案1

得分: 9

程序在主goroutine退出时终止，所以dispatcher()没有时间执行任何操作。你需要在main()中阻塞，直到dispatcher()完成。可以使用通道来实现这一点：

  1. package main
  3. import (
  4.     "fmt"
  5.     "os"
  6.     "bufio"
  7. )
  9. var done = make(chan bool)             // 创建通道
  11. // 加载文件并将它们发送到通道中供mappers读取。
  12. func dispatcher(arr []string,channel chan string) {
  13.     for _,path := range arr {
  14.         file,err := os.Open(path)
  15.         fmt.Println("开始分发 ", path)
  16.         if err != nil {
  17.             fmt.Println(err)
  18.             os.Exit(-1)
  19.         }
  20.         defer file.Close()
  21.         reader := bufio.NewReaderSize(file, 32*10*1024)
  22.         i := 0
  23.         for {
  24.             line,_ := reader.ReadString('\n')
  25.             channel <- line
  26.             i++
  27.             if i%200 == 0 {
  28.                 fmt.Println(i, " 行已解析")
  29.             }
  30.         }
  31.         fmt.Println("完成分发 ", path)
  32.     }
  33. 	done <- true                 // 通知main()完成
  34. }
  36. func main() {
  37.     path := os.Args
  38.     if len(path) < 2 {
  39.         fmt.Println("需要输入文件")
  40.         os.Exit(0)
  41.     }
  42.     channel := make(chan string)
  43.     fmt.Println("在dispatcher之前")
  44.     go dispatcher(path[1:],channel)
  45. 	<-done                 // 等待dispatcher()
  46.     fmt.Println("在dispatcher之后")
  47. }
英文:

Program terminates when main goroutine exits, so that dispatcher() has no time to do anything. You need to block in main() until dispatcher() completes. Channel can be used for this:

  1. package main
  3. import (
  4.     &quot;fmt&quot;
  5.     &quot;os&quot;
  6.     &quot;bufio&quot;
  7. )
  9. var done = make(chan bool)             // create channel
  11. // Load files and send them into a channel for mappers reading.
  12. func dispatcher(arr []string,channel chan string) {
  13.     for _,path := range arr {
  14.         file,err := os.Open(path)
  15.         fmt.Println(&quot;begin to dispatch &quot;, path)
  16.         if err != nil {
  17.             fmt.Println(err)
  18.             os.Exit(-1)
  19.         }
  20.         defer file.Close()
  21.         reader := bufio.NewReaderSize(file, 32*10*1024)
  22.         i := 0
  23.         for {
  24.             line,_ := reader.ReadString(&#39;\n&#39;)
  25.             channel &lt;- line
  26.             i++
  27.             if i%200 == 0 {
  28.                 fmt.Println(i,&quot; lines parsed&quot;)
  29.             }
  30.         }
  31.         fmt.Println(&quot;finish dispatch &quot;, path)
  32.     }
  33. 	done &lt;- true                 // notify main() of completion
  34. }
  36. func main() {
  37.     path := os.Args
  38.     if len(path) &lt; 2 {
  39.         fmt.Println(&quot;Need Input Files&quot;)
  40.         os.Exit(0)
  41.     }
  42.     channel := make(chan string)
  43.     fmt.Println(&quot;before dispatcher&quot;)
  44.     go dispatcher(path[1:],channel)
  45. 	&lt;-done                 // wait for dispatcher()
  46.     fmt.Println(&quot;after dispatcher&quot;)
  47. }

答案2

得分: 2

我修改了你的示例，使其在Go playground上运行，因为那里没有文件I/O；它会在通道上发送随机数。

Victor Deryagin的解释和他建议使用“done”通道是正确的。你遇到死锁的原因是你的goroutine在通道上发送数据，但没有人从中读取，所以程序在这一点上被卡住了。在上面的链接中，我添加了一个消费者goroutine。然后程序按预期并发运行。

请注意，要等待多个goroutine，使用sync.WaitGroup更清晰和更容易。

英文:

I modified your example to run on the Go playground where there's no file I/O; it sends random numbers on the channel instead.

@Victor Deryagin's explanation and his suggestion of using a "done" channel is correct. The reason you get a deadlock is that your goroutine sends on channel, but no one reads from it, so the program is stuck at this point. In the above link I added a consumer goroutine. The program then runs concurrently as intended.

Note that to wait for several goroutines, it's clearer and easier to use sync.WaitGroup.

答案3

得分: 1

在原始问题中有两个问题需要解决。

  1. 在发送完所有数据后，必须关闭通道。在laodData函数中，请在发送完所有数据后使用close(channel)。
  2. 将sync.Waitgroup作为引用传递。你在以下函数的参数中将wg作为值发送... laodData和dispatcher函数。

修复这两个问题将解决死锁问题。你的代码死锁的原因如下：

  • 未关闭发送通道将导致下游通道等待时间过长。
  • 将sync.Waitgroup的参数作为值发送。应该将其作为引用发送，否则将创建一个新的对象副本。
英文:

Two issues needs to be fixed in the original question.

  1. You have to close the channel once you're done sending all the data. In func laodData, please use close(channel) post sending all data.
  2. Pass the sync.Waitgroup as a reference.you are sending wg as a value in the argument to the following functions... laodData and dispatcher functions.

Fixing these two issues will fix your problem of deadlock. The reasons for the deadlock in your code follow:

  • Leaving the sending channel unclosed will cause the downstream channel to wait for prolonged time.
  • sending the argument of sync.Waitgroup as a value . It should be sent as a reference otherwise it will create a new copy of the object which you are sending.

huangapple
  • 本文由 发表于 2012年11月22日 23:28:26
  • 转载请务必保留本文链接：https://go.coder-hub.com/13515846.html
  • go
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定