讲透Go中的并发接收控制结构select

简介: 讲透Go中的并发接收控制结构select

本节源码位置 https://github.com/golang-minibear2333/golang/blob/master/4.concurrent/4.5-select


4.5.1 select与switch


让我们来复习一下switch语句,在switch语句中,会逐个匹配case语句(可以是值也可以是表达式),一个一个的判断过去,直到有符合的语句存在,执行匹配的语句内容后跳出switch。


func demo(number int){
    switch{
        case number >= 90:
        fmt.Println("优秀")
        default:
        fmt.Println("太搓了")
    }
}

而 select 用于处理通道,它的语法与 switch 非常类似。每个 case 语句里必须是一个 channel 操作。它既可以用于 channel 的数据接收,也可以用于 channel 的数据发送。

func foo() {
  chanInt := make(chan int)
  defer close(chanInt)
  go func() {
    select {
    case data, ok := <-chanInt:
      if ok {
        fmt.Println(data)
      }
    default:
      fmt.Println("全部阻塞")
    }
  }()
    time.Sleep(time.Second)
    chanInt <- 1
}


输出1


这是一个简单的接收发送模型。

如果 select 的多个分支都满足条件,则会随机的选取其中一个满足条件的分支。

第6行加上ok是因为上一节讲过,如果不加会导致通道关闭时收到零值。

回忆之前的知识,接收和发送应该在不同的goroutine里。

其次select default子协程,在case都处于阻塞状态时,会直接执行default的内容。导致子协程提前退出,主协程中的写入操作会一直阻塞(等待接收者,接收者已经退出了) 触发死锁

倒数第二行加了sleep 1秒,是因为让select语句提前结束的问题暴露出来。

全部阻塞
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan send]:
main.bar()

select 执行完了,退出了goroutine,而发送才刚刚执行到,没有与其匹配的接收,故死锁。

正确的做法是把接收套在循环里面。

func bar() {
  chanInt := make(chan int)
  defer close(chanInt)
  go func() {
    for {
      select {
          ...
      }
    }
  }()
  chanInt <- 1
}
  • 不再死锁了
  • 假如程序不停止,会出现一个泄露的goroutine,永远的在for循环中无法跳出,此时引入下一节的内容


4.5.2 通知机制


Go 语言总是简单和灵活的,虽然没有针对提供专门的机制来处理退出,但我们可以自己组合

func main() {
  chanInt, done := make(chan int), make(chan struct{})
  defer close(chanInt)
  defer close(done)
  go func() {
    for {
      select {
      case <-chanInt:
      case <-done:
        return
      }
    }
  }()
  done <- struct{}{}
}
  • 没有给chanInt发送任何东西,按理说会阻塞,导致goroutine泄露
  • 但可以使用额外的通道完成协程的退出控制
  • 这种方式还可以做到周期性处理任务,下一节我们再详细讲解


4.5.3 case执行原理


假如case后左边和右边跟了函数,会执行函数,我们来探索一下。


定义A、B函数,作用相同

func A() int {
  fmt.Println("start A")
  time.Sleep(1 * time.Second)
  fmt.Println("end A")
  return 1
}

定义函数lee,请问该函数执行完成耗时多少呢?

func lee() {
  ch, done := make(chan int), make(chan struct{})
  defer close(ch)
  go func() {
    select {
    case ch <- A():
    case ch <- B():
    case <-done:
    }
  }()
  done <- struct{}{}
}

答案是2秒


start A
end A
start B
end B
main.leespend time: 2.003504395s


select扫描是从左到右从上到下的,按这个顺序先求值,如果是函数会先执行函数。

然后立马判断是否可以立即执行(这里是指case是否会因为执行而阻塞)。

所以两个函数都会进入,而且是先进入A再进入B,两个函数都会执行完,所以等待时间会累计。

所以不应该在case判断中放函数。


如果都不会阻塞,此时就会使用一个伪随机的算法,去选中一个case,只要选中了其他就被放弃了。


4.5.4 超时控制


我们来模拟一个更真实点的例子,让程序一段时间超时退出。

定义一个结构体

type Worker struct {
  stream  <-chan int //处理
  timeout time.Duration //超时
  done    chan struct{} //结束信号
}

定义初始化函数

func NewWorker(stream <-chan int, timeout int) *Worker {
  return &Worker{
    stream:  stream,
    timeout: time.Duration(timeout) * time.Second,
    done:    make(chan struct{}),
  }
}

定义超时处理函

func (w *Worker) afterTimeStop() {
  go func() {
    time.Sleep(w.timeout)
    w.done <- struct{}{}
  }()
}


  • 超过时间发送结束信号

接收数据并处理函数

func (w *Worker) Start() {
  w.afterTimeStop()
  for {
    select {
    case data, ok := <-w.stream:
      if !ok {
        return
      }
      fmt.Println(data)
    case <-w.done:
      close(w.done)
      return
    }
  }
}
  • 收到结束信号关闭函数
  • 这样的方法就可以让程序在等待 1 秒后继续执行,而不会因为 ch 读取等待而导致程序停滞。
func main() {
  stream := make(chan int)
  defer close(stream)
  w := NewWorker(stream, 3)
  w.Start()
}

实际3秒到程序运行结束。好在官方已经考虑到这一点,为我们提供了现成的方案。


4.5.5 官方超时方案

go func() {
    t := time.NewTicker(timeout)
    defer t.Stop()
    for {
        select {
        case data := <-chanInt:
            t.Reset(timeout)
        case <-t.C:
        case <-done:
            return
        }
    }
}()


time.NewTicker创建了一个定时器,参数为时间间隔,并返回一个结构体t。

t.C 是一个仅可接收的channel,会根据时间间隔定时执行任务,也可以作为超时任务使用。

t.Reset(timeout) 重置时间,因为select进入一个case,后续的执行会有耗时,所以要重置时间保证时间的精准。

这种方式巧妙地实现了超时处理机制,这种方法不仅简单,在实际项目开发中也是非常实用的。


在生产中,常常把buf积累到一定数量然后flush出去,假如数据产生速度太慢,就要靠定时器定时消费,看下面完整的例子。

func main() {
  chanInt, done := make(chan int), make(chan struct{})
  defer close(chanInt)
  defer close(done)
  go func() {
    ...
  }()
  for i := 0; i < 100; i++ {
    if i%10 == 0 {
      time.Sleep(time.Second)
    }
    chanInt <- 1
  }
  done <- struct{}{}
}

产生100个数,每10个数暂停1秒,用来模拟数据产生速度慢,go func() 内容如下:

go func() {
    timeout := time.Second
    t := time.NewTicker(timeout)
    defer t.Stop()
    buf := make([]int, 0, 5)
    for {
        select {
        case data := <-chanInt:
            t.Reset(timeout)
            if len(buf) < cap(buf) {
                buf = append(buf, data)
            } else {
                go send(buf)
                buf = make([]int, 0, cap(buf))
            }
        case <-t.C:
            if len(buf) > 0 {
                go send(buf)
                buf = make([]int, 0, cap(buf))
            }
        case <-done:
            return
        }
    }
}()


  • 接收到数据时,如果buf满了就进行上报,如果buf没满就追加数据。
  • 假如超时,就直接发送buf防止数据太少一直不发送的情况。
  • 需要在其他case里,Reset超时时间,以校准定时器。


4.5.6 小结


本节介绍了select的用法以及包含的陷阱,我们学会了:


case只针对通道传输阻塞做特殊处理,如果有计算将会先进行计算,所以不应该在case判断中放函数。

扫描是从左到右从上到下的,按这个顺序先求值,如果是函数会先执行函数。如果函数运行时间长,时间会累计。

在case全部阻塞时,会执行default中的内容。

可使用结束信号,让select退出。

延时发送结束信号可以实现超时自动退出的功能。

官方的time包,提供了定时器,可作定时任务,也可作超时控制。


我还写了可热更新的定时器,有兴趣了解的可以看看本节的源码哦。

相关文章
|
10天前
|
人工智能 Go 调度
掌握Go并发:Go语言并发编程深度解析
掌握Go并发:Go语言并发编程深度解析
|
10天前
|
程序员 Go
Golang深入浅出之-Select语句在Go并发编程中的应用
【4月更文挑战第23天】Go语言中的`select`语句是并发编程的关键,用于协调多个通道的读写。它会阻塞直到某个通道操作可行,执行对应的代码块。常见问题包括忘记初始化通道、死锁和忽视`default`分支。要解决这些问题,需确保通道初始化、避免死锁并添加`default`分支以处理无数据可用的情况。理解并妥善处理这些问题能帮助编写更高效、健壮的并发程序。结合使用`context.Context`和定时器等工具,可提升`select`的灵活性和可控性。
30 2
|
10天前
|
SQL Go 数据库
【Go语言专栏】Go语言中的事务处理与并发控制
【4月更文挑战第30天】Go语言在数据库编程中支持事务处理和并发控制,确保ACID属性和多用户环境下的数据完整性。`database/sql`包提供事务管理,如示例所示,通过`Begin()`、`Commit()`和`Rollback()`执行和控制事务。并发控制利用Mutex、WaitGroup和Channel防止数据冲突。结合事务与并发控制,开发者可处理复杂场景,实现高效、可靠的数据库应用。
|
10天前
|
Cloud Native Go 云计算
多范式编程语言Go:并发与静态类型的结合
Go语言是Google于2007年开发的开源编程语言,旨在提高程序开发和部署的效率。它的独特特征在于结合了并发处理与静态类型系统,提供了简洁、高效、并行处理能力的编程体验。本文将探讨Go语言的特点、应用场景以及其在现代软件开发中的优势。
|
10天前
|
安全 Go
Golang深入浅出之-Go语言中的并发安全队列:实现与应用
【5月更文挑战第3天】本文探讨了Go语言中的并发安全队列,它是构建高性能并发系统的基础。文章介绍了两种实现方法:1) 使用`sync.Mutex`保护的简单队列,通过加锁解锁确保数据一致性;2) 使用通道(Channel)实现无锁队列,天生并发安全。同时,文中列举了并发编程中常见的死锁、数据竞争和通道阻塞问题,并给出了避免这些问题的策略,如明确锁边界、使用带缓冲通道、优雅处理关闭以及利用Go标准库。
28 5
|
10天前
|
存储 缓存 安全
Golang深入浅出之-Go语言中的并发安全容器:sync.Map与sync.Pool
Go语言中的`sync.Map`和`sync.Pool`是并发安全的容器。`sync.Map`提供并发安全的键值对存储,适合快速读取和少写入的情况。注意不要直接遍历Map,应使用`Range`方法。`sync.Pool`是对象池,用于缓存可重用对象,减少内存分配。使用时需注意对象生命周期管理和容量控制。在多goroutine环境下,这两个容器能提高性能和稳定性,但需根据场景谨慎使用,避免不当操作导致的问题。
39 4
|
10天前
|
安全 Go 开发者
Golang深入浅出之-Go语言中的CSP模型:深入理解并发哲学
【5月更文挑战第2天】Go语言的并发编程基于CSP模型,强调通过通信共享内存。核心概念是goroutines(轻量级线程)和channels(用于goroutines间安全数据传输)。常见问题包括数据竞争、死锁和goroutine管理。避免策略包括使用同步原语、复用channel和控制并发。示例展示了如何使用channel和`sync.WaitGroup`避免死锁。理解并发原则和正确应用CSP模型是编写高效安全并发程序的关键。
38 4
|
10天前
|
安全 Go 开发者
Golang深入浅出之-Go语言中的CSP模型:深入理解并发哲学
【5月更文挑战第1天】Go语言基于CSP理论,借助goroutines和channels实现独特的并发模型。Goroutine是轻量级线程,通过`go`关键字启动,而channels提供安全的通信机制。文章讨论了数据竞争、死锁和goroutine泄漏等问题及其避免方法,并提供了一个生产者消费者模型的代码示例。理解CSP和妥善处理并发问题对于编写高效、可靠的Go程序至关重要。
28 2
|
10天前
|
设计模式 Go 调度
Golang深入浅出之-Go语言中的并发模式:Pipeline、Worker Pool等
【5月更文挑战第1天】Go语言并发模拟能力强大,Pipeline和Worker Pool是常用设计模式。Pipeline通过多阶段处理实现高效并行,常见问题包括数据竞争和死锁,可借助通道和`select`避免。Worker Pool控制并发数,防止资源消耗,需注意任务分配不均和goroutine泄露,使用缓冲通道和`sync.WaitGroup`解决。理解和实践这些模式是提升Go并发性能的关键。
32 2
|
10天前
|
监控 安全 Go
【Go语言专栏】Go语言中的并发性能分析与优化
【4月更文挑战第30天】Go语言以其卓越的并发性能和简洁语法著称,通过goroutines和channels实现并发。并发性能分析旨在解决竞态条件、死锁和资源争用等问题,以提升多核环境下的程序效率。使用pprof等工具可检测性能瓶颈,优化策略包括减少锁范围、使用无锁数据结构、控制goroutines数量、应用worker pool和优化channel使用。理解并发模型和合理利用并发原语是编写高效并发代码的关键。