讲透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包,提供了定时器,可作定时任务,也可作超时控制。


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

相关文章
|
1月前
|
程序员 Go
go语言中的控制结构
【11月更文挑战第3天】
98 58
|
25天前
|
存储 负载均衡 监控
如何利用Go语言的高效性、并发支持、简洁性和跨平台性等优势,通过合理设计架构、实现负载均衡、构建容错机制、建立监控体系、优化数据存储及实施服务治理等步骤,打造稳定可靠的服务架构。
在数字化时代,构建高可靠性服务架构至关重要。本文探讨了如何利用Go语言的高效性、并发支持、简洁性和跨平台性等优势,通过合理设计架构、实现负载均衡、构建容错机制、建立监控体系、优化数据存储及实施服务治理等步骤,打造稳定可靠的服务架构。
31 1
|
1月前
|
Go 调度 开发者
探索Go语言中的并发模式:goroutine与channel
在本文中,我们将深入探讨Go语言中的核心并发特性——goroutine和channel。不同于传统的并发模型,Go语言的并发机制以其简洁性和高效性著称。本文将通过实际代码示例,展示如何利用goroutine实现轻量级的并发执行,以及如何通过channel安全地在goroutine之间传递数据。摘要部分将概述这些概念,并提示读者本文将提供哪些具体的技术洞见。
|
2月前
|
Java 大数据 Go
Go语言:高效并发的编程新星
【10月更文挑战第21】Go语言:高效并发的编程新星
53 7
|
1月前
|
并行计算 安全 Go
Go语言的并发特性
【10月更文挑战第26天】Go语言的并发特性
17 1
|
2月前
|
安全 Go 调度
探索Go语言的并发模式:协程与通道的协同作用
Go语言以其并发能力闻名于世,而协程(goroutine)和通道(channel)是实现并发的两大利器。本文将深入了解Go语言中协程的轻量级特性,探讨如何利用通道进行协程间的安全通信,并通过实际案例演示如何将这两者结合起来,构建高效且可靠的并发系统。
|
2月前
|
安全 Go 开发者
破译Go语言中的并发模式:从入门到精通
在这篇技术性文章中,我们将跳过常规的摘要模式,直接带你进入Go语言的并发世界。你将不会看到枯燥的介绍,而是一段代码的旅程,从Go的并发基础构建块(goroutine和channel)开始,到高级模式的实践应用,我们共同探索如何高效地使用Go来处理并发任务。准备好,让Go带你飞。
|
2月前
|
安全 Go 调度
探索Go语言的并发之美:goroutine与channel
在这个快节奏的技术时代,Go语言以其简洁的语法和强大的并发能力脱颖而出。本文将带你深入Go语言的并发机制,探索goroutine的轻量级特性和channel的同步通信能力,让你在高并发场景下也能游刃有余。
|
2月前
|
安全 程序员 Go
深入浅出Go语言的并发之道
在本文中,我们将探索Go语言如何优雅地处理并发编程。通过对比传统多线程模型,我们将揭示Go语言独特的goroutine和channel机制是如何简化并发编程,并提高程序的效率和稳定性。本文不涉及复杂的技术术语,而是用通俗易懂的语言,结合生动的比喻,让读者能够轻松理解Go语言并发编程的核心概念。
|
2月前
|
安全 Go 调度
探索Go语言的并发之美:goroutine与channel的实践指南
在本文中,我们将深入探讨Go语言的并发机制,特别是goroutine和channel的使用。通过实际的代码示例,我们将展示如何利用这些工具来构建高效、可扩展的并发程序。我们将讨论goroutine的轻量级特性,channel的同步通信能力,以及它们如何共同简化并发编程的复杂性。