在go中监听多个channel

简介: 在go中监听多个channel

select 关键字


我们可以使用select关键字来同时监听多个goroutine。


package main
import (
    "fmt"
    "time"
)
func main() {
    c1 := make(chan string)
    c2 := make(chan string)
    go func() {
        time.Sleep(1 * time.Second)
        c1 <- time.Now().String()
    }()
    go func() {
        time.Sleep(2 * time.Second)
        c2 <- time.Now().String()
    }()
    for i := 0; i < 2; i++ {
        select {
        case res1 := <-c1:
            fmt.Println("from c1:", res1)
        case res2 := <-c2:
            fmt.Println("from c2:", res2)
        }
    }
}
from c1: 2022-09-04 14:30:39.4469184 -0400 EDT m=+1.000172801
from c2: 2022-09-04 14:30:40.4472699 -0400 EDT m=+2.000524401


上面的代码显示了select关键字是如何工作的:


  • 我们首先创建两个 channel c1和c2来监听。
  • 然后我们生成两个goroutine,分别向c1和c2发送当前时间。
  • 在for循环中,我们创建了一个select语句,并定义了两个 case:第一个 case 是我们何时可以从c1接收,第二个 case 是我们何时可以从c2接收。

你可以看到select语句和switch语句在设计上非常相似。两者都定义了不同的情况,并在满足某种情况时运行相应的代码。另外,我们可以看到select语句是阻塞的。也就是说,它将等待,直到其中一个 case 被满足。


我们为这个循环迭代了两次,因为只有两个goroutine需要监听。更确切地说,每个goroutine都是一个fire-and-forget goroutine,意味着它们在返回之前只向一个通道发送一次。因此,在这段代码中,任何时候都有两个消息的最大值,而我们只需要选择两次。


如果我们不知道工作何时会结束呢?


有时我们不知道有多少个工作。在这种情况下,把 select 语句放在一个while循环里。


package main
import (
    "fmt"
    "math/rand"
    "time"
)
func main() {
    c1 := make(chan string)
    rand.Seed(time.Now().UnixNano())
    for i := 0; i < rand.Intn(10); i++ {
        go func() {
            time.Sleep(1 * time.Second)
            c1 <- time.Now().String()
        }()
    }
    for {
        select {
        case res1 := <-c1:
            fmt.Println("from c1:", res1)
        }
    }
}


因为我们让一个随机数的goroutines运行,所以我们不知道有多少个作业。值得庆幸的是,底部包裹着select语句的for循环将捕获每一个输出。让我们看看如果我们运行这段代码会发生什么。


from c1: 2022-09-04 14:48:47.5145341 -0400 EDT m=+1.000257801
from c1: 2022-09-04 14:48:47.5146126 -0400 EDT m=+1.000336201
from c1: 2022-09-04 14:48:47.5146364 -0400 EDT m=+1.000359901
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan receive]:
main.main()
        /home/jacob/blog/testing/listening-to-multiple-channels-in-go/main.go:22 +0x128
exit status 2


嗯,select 语句按预期收到了三次,但由于死锁,程序出错了。为什么会出现这种情况?


记住,在没有任何条件的情况下,Go中的for循环会永远运行。这意味着 select 语句将永远尝试接收。然而,要运行的作业数量是有限的。即使没有更多的工作,select 语句仍然会尝试接收。


还记得在本系列的第一篇文章中我说过,如果你在发送方没有准备好的时候试图从一个没有缓冲的通道中接收,你的程序就会陷入死锁。这正是我们例子中的情况。


那么我们如何解决这个问题呢?我们可以使用以前文章中涉及的概念的组合:退出通道和WaitGroups。

package main
import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)
func main() {
    c1 := make(chan string)
    exit := make(chan struct{})
    rand.Seed(time.Now().UnixNano())
    var wg sync.WaitGroup
    go func() {
        numJob := rand.Intn(10)
        fmt.Println("number of jobs:", numJob)
        for i := 0; i < numJob; i++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                time.Sleep(1 * time.Second)
                c1 <- time.Now().String()
            }()
        }
        wg.Wait()
        close(exit)
    }()
    for {
        select {
        case res1 := <-c1:
            fmt.Println("from c1:", res1)
        case <-exit:
            return
        }
    }
}
3
from c1: 2022-09-04 15:09:08.6936976 -0400 EDT m=+1.000287801
from c1: 2022-09-04 15:09:08.6937788 -0400 EDT m=+1.000369101
from c1: 2022-09-04 15:09:08.6937949 -0400 EDT m=+1.000385101
  • 我们创建一个 exit channel 和一个WaitGroup。
  • 每次运行的作业数都是随机的。对于numJobs的次数,我们启动goroutines。为了等待作业的完成,我们把它们加到wg中。当一个工作完成后,我们从wg中减去一个。
  • 一旦所有工作完成,我们就关闭exit channel
  • 我们将上述部分包裹在一个goroutine中,因为我们希望所有的部分都是无阻塞的。如果我们不把它包在一个goroutine中,wg.Wait()就会等待,直到作业完成。这将阻塞代码,并且不会让底部的for-select语句运行。
  • 此外,由于c1是一个未缓冲的通道,等待所有的goroutine将消息发送到c1,将导致许多消息被发送到c1而没有for-select语句接收。这就导致了一个死锁,因为当发送者准备好时,接收者还没有准备好。


如何使select成为非阻塞式


select语句默认是阻塞的。我们如何使其成为非阻塞的呢?这很简单--我们只需添加一个 default case。


package main
import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)
func main() {
    ashleyMsg := make(chan string)
    brianMsg := make(chan string)
    exit := make(chan struct{})
    rand.Seed(time.Now().UnixNano())
    var wg sync.WaitGroup
    go func() {
        numJob := rand.Intn(10)
        fmt.Println("number of jobs:", numJob)
        for i := 0; i < numJob; i++ {
            wg.Add(2)
            go func() {
                defer wg.Done()
                time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
                ashleyMsg <- "hi"
            }()
            go func() {
                defer wg.Done()
                time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
                brianMsg <- "what's up"
            }()
        }
        wg.Wait()
        close(exit)
    }()
    for {
        select {
        case res1 := <-ashleyMsg:
            fmt.Println("ashley:", res1)
        case res2 := <-brianMsg:
            fmt.Println("brian:", res2)
        case <-exit:
            fmt.Println("chat ended")
            return
        default:
            fmt.Println("...")
            time.Sleep(time.Millisecond)
        }
    }
}
...
number of jobs: 4
brian: what's up
...
ashley: hi
...
...
brian: what's up
ashley: hi
ashley: hi
brian: what's up
...
...
ashley: hi
...
brian: what's up
...
chat ended


除了蹩脚的对话之外,我们可以看到默认情况下的工作方式。与其等待聊天记录的到来,我们可以在没有频道可以接收的时候做一些事情。在这个例子中,我们只是打印出省略号,但你可以做任何你想做的事情。

相关文章
|
6月前
|
存储 安全 Java
Go 基础数据结构的底层原理(slice,channel,map)
Go 基础数据结构的底层原理(slice,channel,map)
94 0
|
6天前
|
Go 调度 开发者
探索Go语言中的并发模式:goroutine与channel
在本文中,我们将深入探讨Go语言中的核心并发特性——goroutine和channel。不同于传统的并发模型,Go语言的并发机制以其简洁性和高效性著称。本文将通过实际代码示例,展示如何利用goroutine实现轻量级的并发执行,以及如何通过channel安全地在goroutine之间传递数据。摘要部分将概述这些概念,并提示读者本文将提供哪些具体的技术洞见。
|
28天前
|
安全 Go 调度
探索Go语言的并发之美:goroutine与channel
在这个快节奏的技术时代,Go语言以其简洁的语法和强大的并发能力脱颖而出。本文将带你深入Go语言的并发机制,探索goroutine的轻量级特性和channel的同步通信能力,让你在高并发场景下也能游刃有余。
|
1月前
|
存储 安全 Go
探索Go语言的并发模型:Goroutine与Channel
在Go语言的多核处理器时代,传统并发模型已无法满足高效、低延迟的需求。本文深入探讨Go语言的并发处理机制,包括Goroutine的轻量级线程模型和Channel的通信机制,揭示它们如何共同构建出高效、简洁的并发程序。
|
5月前
|
Go
go之channel关闭与广播
go之channel关闭与广播
|
23天前
|
存储 Go 调度
深入理解Go语言的并发模型:goroutine与channel
在这个快速变化的技术世界中,Go语言以其简洁的并发模型脱颖而出。本文将带你穿越Go语言的并发世界,探索goroutine的轻量级特性和channel的同步机制。摘要部分,我们将用一段对话来揭示Go并发模型的魔力,而不是传统的介绍性文字。
|
29天前
|
安全 Go 调度
探索Go语言的并发模型:Goroutine与Channel的魔力
本文深入探讨了Go语言的并发模型,不仅解释了Goroutine的概念和特性,还详细讲解了Channel的用法和它们在并发编程中的重要性。通过实际代码示例,揭示了Go语言如何通过轻量级线程和通信机制来实现高效的并发处理。
|
1月前
|
安全 Go 调度
探索Go语言的并发之美:goroutine与channel的实践指南
在本文中,我们将深入探讨Go语言的并发机制,特别是goroutine和channel的使用。通过实际的代码示例,我们将展示如何利用这些工具来构建高效、可扩展的并发程序。我们将讨论goroutine的轻量级特性,channel的同步通信能力,以及它们如何共同简化并发编程的复杂性。
|
1月前
|
安全 Go 数据处理
掌握Go语言并发:从goroutine到channel
在Go语言的世界中,goroutine和channel是构建高效并发程序的基石。本文将带你一探Go语言并发机制的奥秘,从基础的goroutine创建到channel的同步通信,让你在并发编程的道路上更进一步。
|
3月前
|
消息中间件 Kafka Go
从Go channel中批量读取数据
从Go channel中批量读取数据