go 协程与通道

我就是我 2023-10-09 22:01 52阅读 0赞

#### 阅读目录 ####

*  协程与通道
 *   *  协程 (goroutine)
     *  通道 (channel)
     *   *  1 声明与初始化
         *  2 通信操作符 <-
         *  3 通道阻塞
         *  4 信号量（Semaphore）
         *  5 通道工厂
         *  6 给通道使用 for 循环
         *  7 关闭通道
     *  Select
     *  示例
     *   *  1 惰性生成器
 *  make 与 new 的区别

## 协程与通道 ##

首先来回顾在操作系统中学过的一些概念。

进程 (processes) 是程序执行的基本单位，运行在一个独立的内存地址空间中；

一个进程由多个线程 (threads) 组成，线程的存在是为了能够同时执行多个任务，最大化利用时间，防止产生等待，线程间是共享内存地址空间的。

从 windows 资源管理器看这一点能看的很明白，如下，每个应用程序是一个进程，Typora程序下有两个线程在同时运行。

![在这里插入图片描述][f35ba6a50ebc45049744151d380ec97b.png]  
并发是建立在多线程之上的概念，将CPU的执行时间划分为许多很小的间隔，多个线程不断地切换执行，从上层看起来就像在同时执行一样，但本质上依然是线性的。

并行则是程序在某个特定的事件同时运行在多个CPU上，多核处理器为并行提供了可能。

因此，并发也可能是并行的。

### 协程 (goroutine) ###

Go原生支持并发，依靠的是协程(goroutine)和通道(channel)两个概念。

goroutines 的概念是为了和 processes、threads、coroutines 等概念区别。

其中 coroutines 也叫做协程，而且这才是常规意义下的协程，goroutines 只在 Go 中有效。

coroutines 是比线程更轻的一个概念，只使用很少的内存和资源。  
它对栈进行分隔，从而动态地增加或缩减内存的使用，栈的管理也是自动的，在协程退出后自动释放空间。

协程可以运行在多个线程间，也可以运行在线程内，它的创建廉价到可以在同一地址空间存在100000个。

这一概念也存在于其它语言(C\#, Java等)中，它与 goroutines 的区别在于：

*  Go 协程意味着并行(或者可以以并行的方式部署)，协程一般不是理论上，Go 协程比协程更加强大。
 *  Go 协程通过通道来通信，协程则通过让出与恢复操作来通信。

以一个简单模型来描述 goroutine：  
它是一个和其它协程在同一地址空间并发执行的函数。  
通过在函数或方法名前加上go关键字来创建和运行一个协程，运行结束后安静的退出(没有任何返回值)。

//并行的运行list.Sort，不等待
    go list.Sort()

Go程序中必须含有的 `main()` 函数可以看作一个协程，尽管它没有通过go关键字启动，在程序初始化的过程中 (init()函数运行)，goroutine 也可以运行。

单纯的结束协程的概念是不够具体的，协程需要和通道来配合。

### 通道 (channel) ###

并发编程的困难之处在于实现对共享变量的正确访问，互斥量的方式是复杂的，Go鼓励采用一种不同的方法，

即在通道 (channel) 上传递共享值，如同Unix管道一般，通道用于发送类型化的数据，在任何给定的时间，只有一个协程可以对通道中的数据进行访问，从而完成协程间的通信，也避开了所有由共享内存导致的陷阱。

这种通过通道进行通信的方式保证同步性的同时，数据的所有权也因此被传递。

这一设计理念最终简化为一句话：  
不要通过共享内存来通信，而通过通信来共享内存。

#### 1 声明与初始化 ####

声明通道的基本形式如下，未初始化的通道值为 `nil`。

var identifier chan datatype

通道只能传输一种类型的数据，比如 `chan int` 或 `chan string`，可以是任意类型，包括空接口 `interface{}` 和通道自己。

和 map 相同，通道也是引用类型，因此使用 make 进行初始化，可以指定第二个参数用来指定缓冲区的大小，即通道可容纳的数据个数，这一个值默认是 0，意思是无缓冲，无缓冲的通道将通信、值的交换、同步三者结合，保证两个协程的计算处于已知状态。

var ci chan string
    // 无缓冲的整数通道
    ci = make(chan string)
    // 无缓冲的整数通道
    cj := make(chan int, 0)
    // 指向文件的指针的缓冲通道
    cs := make(chan *os.File, 100)

#### 2 通信操作符 <- ####

操作符直观的表示数据的传输：信息按照箭头方向流动。

流向通道(发送)用 `ch <- int1` 表示，意为利用通道 ch 发送变量 int1。

从通道流出(接收)用 `int2 = <- ch` 表示，意为变量 int2 从通道 ch 接收数据，如果 int2 没有声明过，可以使用 `int2 := <- ch`。

`<- ch` 则用于表示丢弃当前值，获取通道的下一个值，可以用来验证，如：

if <- ch != 1000 {
        
        ...
    }

为了可读性通道的命名通常以 ch 开头或者包含 chan。  
通道的发送和接收都是原子操作，总是互不干扰的完成。  
下面的示例展示了通信操作符的使用。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func main() {
        
    	ch := make(chan string)
    
    	go sendData(ch)
    	go getData(ch)
    
    	// 当前程序暂停 1 秒钟
    	time.Sleep(1e9)
    }
    
    func sendData(ch chan string) {
        
    	ch <- "Washington"
    	ch <- "Tripoli"
    	ch <- "London"
    	ch <- "Beijing"
    	ch <- "Tokyo"
    }
    
    func getData(ch chan string) {
        
    	var input string
    	// time.Sleep(2e9)
    	for {
        
    		input = <-ch
    		fmt.Printf("%s ", input)
    	}
    }
    //Output:
    Washington Tripoli London Beijing tokyo

如果 2 个协程需要通信，必须给它们同一个通道作为参数。

上例中 main() 函数中启动了两个协程：  
sendData() 通过通道 ch 发送了 5 个字符串，  
getData() 按顺序接收它们并打印出来。

一些同步的细节如下：

1 main() 等待了 1 秒让两个协程完成，如果不这样(注释掉time.Sleep(1e9))，sendData() 就没有机会输出。

2 getData() 使用了无限循环：它随着 sendData() 的发送完成和 ch 变空也结束了。

3 如果我们移除一个或所有 go 关键字，程序无法运行，Go 运行时会抛出 panic。这是因为运行时(runtime)会检查所有的协程是否在等待什么东西(从通道读取或写入某个通道)，这意味着陷入死锁，程序无法继续执行。

通道的发送和接收顺序是无法预知的，如果使用打印状态来输出，由于两者间的时间延迟，打印的顺序和真实发生的顺序是不同的。

#### 3 通道阻塞 ####

前面提到默认情况下通信是同步且无缓冲的，因此通道的发送 `/` 接收操作在对方准备好之前是阻塞的。

1 对于同一个通道，发送操作（协程或者函数中的），在接收者准备好之前是阻塞的：  
如果 ch 中的数据无人接收，就无法再给通道传入其他数据。

2 对于同一个通道，接收操作在发送者可用之前是阻塞的（协程或函数中的）

下例中的协程在无限循环中不断地给通道发送数据，但由于没有接收者，只输出了数字 0。

package main
    
    import "fmt"
    
    func main() {
        
    	ch1 := make(chan int)
    	go pump(ch1)       // pump hangs
    	fmt.Println(<-ch1) // prints only 0
    }
    
    func pump(ch chan int) {
        
    	for i := 0; ; i++ {
        
    		ch <- i
    	}
    }
    //Output:0

定义一个新的协程用来接收通道值可以持续输出。

package main
    
    import "fmt"
    
    func main() {
        
    	ch1 := make(chan int)
    	go pump(ch1)
    	go suck(ch1)
    	time.Sleep(1e9)
    }
    
    func pump(ch chan int) {
        
    	for i := 0; ; i++ {
        
    		ch <- i
    	}
    }
    
    func suck(ch chan int) {
        
    	for {
        
    		fmt.Println(<-ch)
    	}
    }

#### 4 信号量（Semaphore） ####

Go 语言中，信号量（Semaphore）通常用于限制并发访问资源的数量。

Semaphore 是一种计数器对象，可以用于控制同时访问共享资源的线程数量。

Semaphore 通常具有两个基本操作：

*  Acquire 操作会尝试获取一个许可（或信号），如果当前可用的许可数量为 0，则会被阻塞等待，直到有许可可用。
 *  Release 操作会释放一个许可，使得其他被阻塞的线程可以获取该许可。

Go 语言中可以使用 sync.Mutex 和 sync.Cond 实现 Semaphore。

Go 语言中，sync.Cond 是一个条件变量对象，用于在多个 goroutine 之间进行通信。  
它通常与互斥锁（sync.Mutex）一起使用，以实现线程间同步。

当一个 goroutine 想要等待另一个 goroutine 的某个条件满足时，它可以调用 Wait() 方法将自己阻塞，等待条件变量的通知。

当条件满足时，通知可以通过 Signal() 或 Broadcast() 方法发出，以唤醒等待的 goroutine。

下面是一个使用 sync.Mutex 和 sync.Cond 实现 Semaphore 的示例代码：

package main
    
    import (
    	"fmt"
    	"sync"
    )
    
    type Semaphore struct {
        
    	// 当前可用的许可数量
    	count int
    	// 条件变量对象，用于等待和唤醒 goroutine
    	cond *sync.Cond
    }
    
    // 创建许可数量
    func NewSemaphore(count int) *Semaphore {
        
    	return &Semaphore{
        
    		count: count,
    		cond:  sync.NewCond(&sync.Mutex{
        }),
    	}
    }
    
    // 操作许可
    func (s *Semaphore) Acquire() {
        
    	// 获取条件变量对应的锁
    	s.cond.L.Lock()
    	defer s.cond.L.Unlock()
    
    	// 如果当前没有可用的许可，就等待条件变量
    	for s.count <= 0 {
        
    		// 释放锁并等待条件变量满足
    		s.cond.Wait()
    	}
    	s.count-- // 获取许可，许可数量减一
    }
    
    // 释放许可
    func (s *Semaphore) Release() {
        
    	s.cond.L.Lock()
    	defer s.cond.L.Unlock()
    
    	// 释放许可，许可数量加一
    	s.count++
    	// 唤醒等待条件变量的 goroutine 中的一个
    	s.cond.Signal()
    }
    
    func main() {
        
    	sem := NewSemaphore(3)
    	// 协调多个 goroutine 的执行
    	var wg sync.WaitGroup
    	for i := 0; i < 5; i++ {
        
    		wg.Add(1) // 增加计数器的值
    		go func(id int) {
        
    			sem.Acquire()
    			defer sem.Release()
    			fmt.Printf("Goroutine %d acquired 获取许可 semaphore\n", id)
    			// do some work
    			fmt.Printf("Goroutine %d released 释放许可 semaphore\n", id)
    			wg.Done() //减少计数器的值
    		}(i)
    	}
    
    	wg.Wait() //等待计数器归零
    	fmt.Println("All goroutines have finished")
    }

root@debiancc:~/www/test# go run test.go 
    Goroutine 4 acquired 获取许可 semaphore
    Goroutine 4 released 释放许可 semaphore
    Goroutine 2 acquired 获取许可 semaphore
    Goroutine 2 released 释放许可 semaphore
    Goroutine 3 acquired 获取许可 semaphore
    Goroutine 3 released 释放许可 semaphore
    Goroutine 0 acquired 获取许可 semaphore
    Goroutine 0 released 释放许可 semaphore
    Goroutine 1 acquired 获取许可 semaphore
    Goroutine 1 released 释放许可 semaphore
    All goroutines have finished
    root@debiancc:~/www/test#

#### 5 通道工厂 ####

编程时常用一种通道工厂的模式，即不将通道作为参数传递给协程，而是用函数来生成一个通道并返回。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func main() {
        
    	stream := pump()
    	go suck(stream)
    	time.Sleep(1e9)
    }
    
    func pump() chan int {
        
    	ch := make(chan int)
    	go func() {
        
    		for i := 0; ; i++ {
        
    			ch <- i
    		}
    	}()
    	return ch
    }
    
    func suck(ch chan int) {
        
    	for {
        
    		fmt.Println(<-ch)
    	}
    }

在一秒钟内不断的执行执行流入流出。

#### 6 给通道使用 for 循环 ####

for 循环的 range 语句可以用在通道 ch 上，便可以从通道中获取值，像这样：

for v := range ch {
        
    	fmt.Printf("The value is %v\n", v)
    }

这样的使用依然必须和通道的写入和关闭相配合， 不能单独存在。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func main() {
        
    	suck(pump())
    	time.Sleep(1e9)
    }
    
    func pump() chan int {
        
    	ch := make(chan int)
    	go func() {
        
    		for i := 0; ; i++ {
        
    			ch <- i
    		}
    	}()
    	return ch
    }
    
    func suck(ch chan int) {
        
    	go func() {
        
    		for v := range ch {
        
    			fmt.Println(v)
    		}
    	}()
    }

#### 7 关闭通道 ####

通道可以被显式的关闭，不过只有发送者才需要关闭通道，接收者永远不需要。

ch := make(chan float64)
    defer close(ch)

测试通道是否关闭则可以使用 ok 操作符。

v, ok := <-ch   // ok is true if v received value

一个完整的例子

package main
    
    import "fmt"
    
    func main() {
        
    	ch := make(chan string)
    	go sendData(ch)
    	getData(ch)
    }
    
    func sendData(ch chan string) {
        
    	ch <- "Washington"
    	ch <- "Tripoli"
    	ch <- "London"
    	ch <- "Beijing"
    	ch <- "Tokio"
    	close(ch)
    }
    
    func getData(ch chan string) {
        
    	for {
        
    		input, open := <-ch
    		if !open {
        
    			break
    		}
    		fmt.Printf("%s ", input)
    	}
    }

root@debiancc:~/www/test# go run test.go 
    Washington Tripoli London Beijing Tokio
    root@debiancc:~/www/test#

但是使用 for-range 读取通道是更好的办法，因为这会自动检测通道是否关闭。

for input := range ch {
        
      	process(input)
    }

### Select ###

从不同的并发执行的协程中获取值可以通过关键字 select 来完成，它和 switch 控制语句非常相似，其行为像是“你准备好了吗”的轮询机制；

select 监听进入通道的数据，也可以是用通道发送值的时候。

select {
        
    case u:= <- ch1:
            ...
    case v:= <- ch2:
            ...
            ...
    default: // no value ready to be received
            ...
    }

select 做的事情是：选择处理列出的多个通信情况中的一个。

*  如果都阻塞了，会等待直到其中一个可以处理。
 *  如果多个可以处理，随机选择一个。
 *  如果没有通道操作可以处理并且写了 default 语句，它就会执行：default 永远是可运行的（这就是准备好了，可以执行）。

使用 default 可以保证发送不被阻塞，但没有 default 的监听模式也可能被使用，通过 break 语句退出循环。

一个完整的例子：

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func main() {
        
    	ch1 := make(chan int)
    	ch2 := make(chan int)
    
    	go pump1(ch1)
    	go pump2(ch2)
    	go suck(ch1, ch2)
    
    	time.Sleep(1e9)
    }
    
    func pump1(ch chan int) {
        
    	for i := 0; ; i++ {
        
    		ch <- i * 2
    	}
    }
    
    func pump2(ch chan int) {
        
    	for i := 0; ; i++ {
        
    		ch <- i + 5
    	}
    }
    
    func suck(ch1, ch2 chan int) {
        
    	for {
        
    		select {
        
    		case v := <-ch1:
    			fmt.Printf("Received on channel 1: %d\n", v)
    		case v := <-ch2:
    			fmt.Printf("Received on channel 2: %d\n", v)
    		}
    	}
    }

有 2 个通道 ch1 和 ch2，三个协程 pump1()、pump2() 和 suck()。

在无限循环中，ch1 和 ch2 通过 pump1() 和 pump2() 填充整数；

suck() 也在无限循环中轮询输入，

通过 select 语句获取 ch1 和 ch2 的整数并输出。

选择哪一个 case 取决于哪一个通道收到了信息。

程序在 main 执行 1 秒后结束。

### 示例 ###

#### 1 惰性生成器 ####

生成器是指当被调用时返回一个序列中下一个值的函数，

例如：

generateInteger() => 0
    generateInteger() => 1
    generateInteger() => 2
    ....

生成器每次返回的是序列中下一个值而非整个序列；

这种特性也称之为惰性求值：  
只在你需要时进行求值，同时保留相关变量资源（内存和cpu）：

这是一项在需要时对表达式进行求值的技术。

例如，生成一个无限数量的偶数序列：  
要产生这样一个序列并且在一个一个的使用可能会很困难，而且内存会溢出！但是一个含有通道和go协程的函数能轻易实现这个需求。

下例中实现了一个使用 int 型通道来实现的生成器。

通道被命名为 yield 和 resume，这些词经常在协程代码中使用。

package main
    
    import (
    	"fmt"
    )
    
    var resume chan int
    
    func integers() chan int {
        
    	yield := make(chan int)
    	count := 0
    	go func() {
        
    		for {
        
    			yield <- count
    			count++
    		}
    	}()
    	return yield
    }
    
    func generateInteger() int {
        
    	return <-resume
    }
    
    func main() {
        
    	resume = integers()
    	fmt.Println(generateInteger()) //=> 0
    	fmt.Println(generateInteger()) //=> 1
    	fmt.Println(generateInteger()) //=> 2
    }

## make 与 new 的区别 ##

在 Go 语言中，make 和 new 都可以用来分配内存，但是它们的使用场景和行为有所不同。

1 make 用于创建切片、映射和管道等引用类型（slices, maps, and channels），并且返回一个已经初始化后的引用类型（slice、map 或 channel）的值。这些类型都是对底层数据结构的引用，需要初始化后才能使用。

示例：

// 创建一个长度为 5，容量为 10 的整型切片
    slice := make([]int, 5, 10) 
    
    // 创建一个字符串类型的管道
    channel := make(chan string)

2 new 用于分配值类型（struct、int、float64 等）的内存空间，并返回一个指向该类型的指针。  
分配的内存会被置为零值，即该类型的默认值。

示例：

var ptr *int
    
    // 分配一个整型的内存空间，并将 ptr 指向该空间
    ptr = new(int)

总的来说，make 和 new 的主要区别在于：  
make 只能用于引用类型的初始化，并且返回已经初始化后的值；  
而 new 则用于分配值类型的内存空间，并返回指向该类型的指针。

> `slice := new([]int, 5, 10)` 这样可以吗？

不可以。

在 Go 语言中，new 函数用于分配值类型（struct、int、float64 等）的内存空间，并返回一个指向该类型的指针。