Go 编程实例【Channel】

淩亂°似流年 2023-10-09 20:41 26阅读 0赞

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

*  Channel
 *   *  定义 Channel
     *  无缓冲 Channel
     *  通过 goroutine+channel 计算数组之和
 *  缓冲 Channel
 *   *  非缓冲 channel 和缓冲 channel 的对比
 *  关闭 channel
 *  遍历 Channel
 *  通过 select 操作 channel
 *  goroutine+channel 统计文本文件中每个单词的数量
 *   *  解决 goroutine 中同时读写同一个 map 时发生冲突

## Channel ##

Channel 是 Go 语言中一种用于在 Goroutine 之间传递数据的机制。

Channel 通过通信实现共享内存，可以安全地传递数据，避免了多个 Goroutine 访问共享内存时出现的竞争和死锁问题。

Channel 可以是有缓冲或无缓冲的。

无缓冲的 Channel，也称为同步 Channel，发送操作和接收操作必须同时准备就绪，否则会被阻塞。

有缓冲的 Channel，也称为异步 Channel，发送操作会在 Channel 缓冲区未满的情况下立即返回，接收操作也会在 Channel 缓冲区不为空的情况下立即返回，否则会被阻塞。

### 定义 Channel ###

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    /*
    定义 channel,
    channel 是带有类型的管道，
    可以通过信道操作符 <- 来发送或者接收值
    */
    func main() {
        
    	// 信道在使用前必须通过内建函数 make 来创建
    	/*
    	   make(chan T,size)
    	   标识用内建函数 make 来创建
    	   一个T类型的缓冲大小为 size 的 channel
    	*/
    	/*
    	   如下: make(chan int) 用内建函数 make
    	   来创建 一个 int 类型的缓冲大小为 0 的 channel
    	*/
    	c := make(chan int)
    
    	go func() {
        
    		// 从 c 接收值并赋予 num
    		num := <-c
    		fmt.Printf("recover:%d\n", num)
    	}()
    
    	// 将 1 发送至信道 c
    	c <- 1
    	<-time.After(time.Second * 3)
    	fmt.Println("return")
    }

root@192:~/www/test# go run main.go
    recover:1
    return
    root@192:~/www/test#

首先通过 make 函数创建了一个无缓冲的 int 类型的 Channel c，即：`c := make(chan int)`。

然后通过 go 关键字定义了一个匿名的 Goroutine，用于从 Channel c 中接收数据。

匿名 Goroutine 中，使用 <- 语法从 Channel c 中接收值，并将其赋值给变量 num。接收完值后，使用 fmt.Printf 打印出接收到的值。

接着，在 main函数 中，使用 <- 语法将整数值 1 发送到 Channel c 中，即：c <- 1。

最后，为了保证 Goroutine 有足够的时间去接收 Channel 中的值，通过 <-time.After(time.Second \* 3) 等待 3 秒钟之后，打印出 “return”。如果将 <-time.After(time.Second \* 3) 去掉，那么程序可能在打印 “return” 之前就结束了，因为 Goroutine 没有足够的时间去接收 Channel 中的值。

### 无缓冲 Channel ###

无缓冲的 Channel通过定义：

make(chan T)

**在无缓冲的 Channel 中，发送和接收操作是同步的。**

如果一个 Goroutine 向一个无缓冲的 Channel 发送数据，它将一直阻塞，直到另一个 Goroutine 从该 Channel 中接收到数据。

同样地，如果一个 Goroutine 从一个无缓冲的 Channel 中接收数据，它将一直阻塞，直到另一个 Goroutine 向该 Channel 中发送数据。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    // 发送端和接收端的阻塞问题
    /*
    发送端在没有准备好之前会阻塞,
    同样接收端在发送端没有准备好之前会阻塞
    */ 
    func main() {
        
    	c := make(chan string)
    
    	go func() {
        
    		<-time.After(time.Second * 10)
    		fmt.Println("发送端准备好了 send: ping")
    		c <- "ping" // 发送
    	}()
    
    	// 发送端10s后才准备好，所以阻塞在当前位置
    	fmt.Println("阻塞在当前位置，发送端发送数据后才继续执行")
    	num := <-c
    	fmt.Printf("recover: %s\n", num)
    }

上面代码创建了一个无缓冲的字符串类型的 Channel c，然后启动了一个新的 Goroutine，该 Goroutine 会在 10 秒后发送一个字符串 “ping” 到 Channel c 中。

在主 main 中，接收操作 <-c 会阻塞，直到有值从 Channel c 中被接收到为止。

因为发送端需要 10 秒后才会发送数据，所以接收端会在 <-c 处阻塞 10 秒。

接收到 “ping” 后，主 main 继续执行，输出 “recover: ping”。

root@192:~/www/test# go run main.go
    阻塞在当前位置，发送端发送数据后才继续执行
    发送端准备好了 send: ping
    recover: ping
    root@192:~/www/test#

### 通过 goroutine+channel 计算数组之和 ###

package main
    
    import "fmt"
    
    /*
    对切片中的数进行求和，将任务分配给两个 Go 程。
    一旦两个 Go 协程完成了它们的计算，它就能算出最终的结果。
    */
    
    // sum 求和函数
    func sum(s []int, c chan int) {
        
    	ans := 0
    	for _, v := range s {
        
    		ans += v
    	}
    	c <- ans // 将和送入 c
    }
    
    func main() {
        
    	s := []int{
        1, 1, 1, 1, 1, 2, 2, 2, 2, 2}
    
    	c := make(chan int)
    	go sum(s[:len(s)/2], c)
    	go sum(s[len(s)/2:], c)
    	x, y := <-c, <-c // 从 c 中接收
    
    	fmt.Println(x, y, x+y)
    }

root@192:~/www/test# go run main.go
    10 5 15
    root@192:~/www/test#

## 缓冲 Channel ##

缓冲channel定义：`make(chan T,size)`

缓冲 Channel 是带有缓冲区的 Channel，创建时需要指定缓冲区大小，例如 `make(chan int, 10)` 创建了一个缓冲区大小为 10 的整型 Channel。

缓冲 Channel 中, 当缓冲区未满时，发送操作是非阻塞的，如果缓冲区已满，则发送操作会阻塞，直到有一个接收操作接收了一个值, 才能继续发送。

当缓冲区非空时，接收操作是非阻塞的，如果缓冲区为空，则接收操作会阻塞，直到有一个发送操作发送了一个值。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func producer(c chan int, n int) {
        
    	for i := 0; i < n; i++ {
        
    		c <- i
    		fmt.Printf("producer sent: %d\n", i)
    	}
    	close(c)
    }
    
    func consumer(c chan int) {
        
    	for {
        
    		num, ok := <-c
    		if !ok {
        
    			fmt.Println("consumer closed")
    			return
    		}
    		fmt.Printf("consumer received: %d\n", num)
    	}
    }
    
    func main() {
        
    	c := make(chan int, 5)
    	go producer(c, 10)
    	go consumer(c)
    	time.Sleep(time.Second * 1)
    	fmt.Println("main exited")
    }

root@192:~/www/test# go run main.go
    producer sent: 0
    producer sent: 1
    producer sent: 2
    producer sent: 3
    producer sent: 4
    producer sent: 5
    consumer received: 0
    consumer received: 1
    consumer received: 2
    consumer received: 3
    consumer received: 4
    consumer received: 5
    consumer received: 6
    producer sent: 6
    producer sent: 7
    producer sent: 8
    producer sent: 9
    consumer received: 7
    consumer received: 8
    consumer received: 9
    consumer closed
    main exited
    root@192:~/www/test#

在上面代码中，我们创建了一个缓冲区大小为 5 的整型 Channel，生产者向 Channel 中发送了 10 个整数，

消费者从 Channel 中接收这些整数，并将它们打印出来。

由于缓冲区大小为 5，因此生产者只有在 Channel 中有 5 个或更少的元素时才会被阻塞。

在该示例中，由于消费者从 Channel 中接收元素的速度比生产者发送元素的速度快，因此生产者最终会被阻塞，直到消费者接收完所有的元素并关闭 Channel。

> 需要注意的是，当 Channel 被关闭后，仍然可以从 Channel 中接收剩余的元素，  
> 但不能再向 Channel 中发送任何元素。

因此，在消费者函数中，我们使用了 for 循环和 ok 标志来检查 Channel 是否已经被关闭。

### 非缓冲 channel 和缓冲 channel 的对比 ###

package main
    
    import "fmt"
    
    // 不带缓冲的 channel
    func NoBufferChan() {
        
    	ch := make(chan int)
    	ch <- 1
    	/*
    	   被阻塞,
    	   执行报错 fatal error: all goroutines are asleep - deadlock!
    	*/
    	fmt.Println(<-ch)
    }
    
    // 带缓冲的 channel
    func BufferChan() {
        
    	// channel 有缓冲、是非阻塞的，直到写满 cap 个元素后才阻塞
    	ch := make(chan int, 1)
    	ch <- 1
    	fmt.Println(<-ch)
    }
    
    func main() {
        
    	NoBufferChan()
    	// BufferChan()
    }

`NoBufferChan()` 报错因为发送操作和接收操作必须同时发生在不同的 goroutine 中，否则程序将会发生死锁。

因为不带缓冲的 channel 是同步的，如果在没有接收者的情况下进行发送操作，发送者将会一直等待直到有接收者为止。

如果没有接收者，发送者将会一直阻塞，直到程序崩溃。

因此，我们需要确保在发送值之前有接收者等待接收这些值，或者使用带缓冲的 channel，这样发送者将不会一直阻塞等待接收者。

## 关闭 channel ##

Close 函数可以用于关闭 Channel，关闭一个 channel 后，可以从中读取数据不过读取的数据全是当前channel 类型的零值，但不能向这个 channel 写入数据会发送 panic。

package main
    
    func main() {
        
      ch := make(chan bool)
      close(ch)
      fmt.Println(<- ch)
      //ch <- true // panic: send on closed channel
    }

<table> 
 <tbody> 
  <tr> 
   <th>操作</th> 
   <th>一个零值nil通道</th> 
   <th>一个非零值但已关闭的通道</th> 
   <th>一个非零值且尚未关闭的通道</th> 
  </tr> 
  <tr> 
   <td>关闭</td> 
   <td>产生恐慌</td> 
   <td>产生恐慌</td> 
   <td>成功关闭</td> 
  </tr> 
  <tr> 
   <td>发送数据</td> 
   <td>永久阻塞</td> 
   <td>产生恐慌</td> 
   <td>阻塞或者成功发送</td> 
  </tr> 
  <tr> 
   <td>接收数据</td> 
   <td>永久阻塞</td> 
   <td>永不阻塞</td> 
   <td>阻塞或者成功接收</td> 
  </tr> 
 </tbody> 
</table>

## 遍历 Channel ##

可以通过 range 持续读取 channel，直到 channel 关闭。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    // 通过 range 遍历 channel, 并通过关闭 channel 来退出循环
    /*
    复制一个 channel 或用于函数参数传递时,
    只是拷贝了一个 channel 的引用,
    因此调用者和被调用者将引用同一个channel对象。
    */
    func genNum(c chan int) {
        
    	for i := 0; i < 10; i++ {
        
    		c <- i
    		time.Sleep(1 * time.Second)
    	}
    	// 发送者可通过 close 关闭一个信道来表示没有需要发送的值了
    	close(c)
    }
    
    func main() {
        
    	c := make(chan int, 10)
    	go genNum(c)
    
    	/*
    	   循环 for v := range c 会不断从信道接收值，直到它被关闭。
    	   并且只有发送者才能关闭信道，而接收者不能,
    	   向一个已经关闭的信道发送数据会引发程序恐慌（panic）
    	*/
    	for v := range c {
        
    		fmt.Println("receive:", v)
    	}
    
    	/*
    		接收者可以通过 v,ok := <- c 表达式接收第二个参数来测试信道是否被关闭：
    		若没有值可以接收且信道已被关闭,那么 v 为对应类型零值,ok 为 false
    	*/
    	v, ok := <-c
    	fmt.Printf("value:%d, ok:%t\n", v, ok)
    
    	fmt.Println("close")
    }

## 通过 select 操作 channel ##

通过 select-case 可以选择一个准备好数据 channel 执行，会从这个 channel 中读取或写入数据。

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    // 通过 channel+select 控制 goroutine 退出
    func genNum(c, quit chan int) {
        
    	for i := 0; ; i++ {
        
    		/*
    			select 可以等待多个通信操作
    			select 会阻塞等待可执行分支。
    			当多个分支都准备好时会随机选择一个执行。
    		*/
    		select {
        
    		case <-quit:
    			// 发送者可通过 close 关闭一个信道来表示没有需要发送的值了。
    			close(c)
    			return
    		default:
    			/*等同于 switch 的 default。
    			当所有case都阻塞时如果有default则，
    			执行default*/
    			c <- i
    			time.Sleep(1 * time.Second)
    		}
    	}
    }
    
    func main() {
        
    	c := make(chan int)
    	quit := make(chan int)
    	go genNum(c, quit)
    
    	/*循环 for v := range c 会不断从信道接收值，直到它被关闭。
    	并且只有发送者才能关闭信道，而接收者不能。
    	向一个已经关闭的信道发送数据会引发程序恐慌（panic）。*/
    	for i := 0; i < 10; i++ {
        
    		fmt.Println("receive:", <-c)
    	}
    
    	// 通知 genNum() 退出
    	quit <- 1
    
    	/*
    		接收者可以通过 v,ok := <- c 表达式第二个参数来测试信道是否被关闭：
    		若没有值可以接收且信道已被关闭，那么在执行完。
    	*/
    	v, ok := <-c
    	fmt.Printf("value:%d, ok:%t\n", v, ok)
    
    	fmt.Println("close")
    }

root@192:~/www/test# go run main.go
    receive: 0
    receive: 1
    receive: 2
    receive: 3
    receive: 4
    receive: 5
    receive: 6
    receive: 7
    receive: 8
    receive: 9
    value:0, ok:false
    close
    root@192:~/www/test#

## goroutine+channel 统计文本文件中每个单词的数量 ##

package main
    
    import (
    	"bufio"
    	"fmt"
    	"log"
    	"os"
    	"strings"
    )
    
    func main() {
        
    	// 打开文件
    	file, err := os.Open("test.txt")
    	if err != nil {
        
    		log.Fatal(err)
    	}
    	defer file.Close()
    
    	// 创建一个 map，用于存储每个单词的数量
    	wordCount := make(map[string]int)
    
    	// 创建一个 channel，用于在不同的 goroutine 中传递数据
    	words := make(chan string)
    
    	// 启动一个 goroutine 读取文件并将每个单词发送到 channel 中
    	go func() {
        
    		/*创建一个新的 Scanner 对象，用来读取文件 file。
    		Scanner 对象提供了一种方便的方式来逐行或逐词读取文件内容。
    		需要注意的是，file 参数需要是一个 *os.File 类型的文件对象，
    		该对象可以通过 os.Open 或 os.Create 等函数创建。*/
    		scanner := bufio.NewScanner(file)
    		/*设置 Scanner 对象的分割函数，这里我们使用 bufio.ScanWords 函数来实现按照单词进行分割。
    		具体来说，bufio.ScanWords 函数会将每个连续的非空白字符序列作为一个单词，
    		并将其作为一个字符串返回。在本例中，我们将使用 ScanWords 来逐个读取文件中的单词。*/
    		scanner.Split(bufio.ScanWords)
    		/*读取文件中的下一个单词或行，并将其保存在 Scanner 对象的缓冲区中。
    		如果成功读取到下一个单词或行，则返回 true，否则返回 false。
    		可以通过调用 scanner.Text() 方法来获取缓冲区中最后一次读取的内容。*/
    		for scanner.Scan() {
        
    			word := strings.ToLower(scanner.Text())
    			words <- word
    		}
    		close(words) // 关闭 channel
    	}()
    
    	// 启动多个 goroutine 统计单词数量
    	for i := 0; i < 3; i++ {
        
    		go func() {
        
    			for word := range words {
        
    				wordCount[word]++
    			}
    		}()
    	}
    
    	// 等待所有的 goroutine 完成并输出结果
    	for i := 0; i < 3; i++ {
        
    		<-words // 等待所有的单词被处理完
    	}
    	fmt.Println(wordCount)
    }

map[again.:1 away:1 brown:3 dog:1 dog.:2 fox:3 from:1 jumps:2 lazy:3 over:2 quick:3 runs:1 the:5]
    root@192:~/www/test# go run main.go
    map[again.:1 away:1 brown:3 dog:1 dog.:2 fox:3 from:1 jumps:2 lazy:3 over:2 quick:3 runs:1 the:5]
    root@192:~/www/test#

在这个示例代码中，我们首先打开文本文件，并创建了一个 map 用于存储每个单词的数量。

然后，我们创建了一个 channel 用于在不同的 goroutine 中传递数据。

接下来，我们启动一个 goroutine 读取文件并将每个单词发送到 channel 中。

在这个 goroutine 中，我们使用 bufio 包中的 Scanner 类型按单词分割文件，并将每个单词转换为小写字母，然后将其发送到 channel 中。

最后，我们关闭 channel。

接下来，我们启动多个 goroutine 统计单词数量。  
在这些 goroutine 中，我们使用 range 关键字从 channel 中接收每个单词，并将其存储在 map 中。由于 map 是 Go 语言中的一个并发安全的数据结构，因此多个 goroutine 可以同时更新它，而不会发生数据竞争的情况。

最后，我们在主 goroutine 中等待所有的 goroutine 完成并输出结果。

在这个示例代码中，我们使用一个 for 循环来等待所有的单词被处理完。  
由于每个 goroutine 都会从 channel 中接收到一个值，因此我们需要等待所有的 goroutine 从 channel 中接收到相同数量的值，才能确保所有的单词都被处理完。

最后，我们输出了每个单词的数量。

test.txt

The quick brown fox jumps over the lazy dog.
    The quick brown fox jumps over the lazy dog again.
    The quick brown fox runs away from the lazy dog.

Go 语言中，map 是一种可以在多个 goroutine 中同时读写的数据结构，且不需要额外的同步操作即可保证并发安全。具体来说，当多个 goroutine 并发读写同一个 map 时，map 内部会自动处理并发访问的问题，保证不会发生数据竞争等并发安全问题。

在实现上，map 使用了一些复杂的技术来实现并发安全，例如分段锁、并发写时复制（copy-on-write）等。这些技术可以有效地减小锁的粒度，提高并发访问的效率，同时保证并发访问的正确性。

> 需要注意的是，虽然 map 在实现上是并发安全的，但在实际使用中仍需注意一些细节问题。  
> 例如在多个 goroutine 中同时读写同一个 map 时可能会发生冲突，需要进行合适的同步操作，否则会导致数据错误或竞态条件等问题。

### 解决 goroutine 中同时读写同一个 map 时发生冲突 ###

package main
    
    import (
    	"bufio"
    	"fmt"
    	"os"
    	"sync"
    )
    
    func countWords(filename string, wg *sync.WaitGroup, mu *sync.Mutex, wordCountMap map[string]int) {
        
    	defer wg.Done()
    
    	file, err := os.Open(filename)
    	if err != nil {
        
    		fmt.Println("error opening file:", err)
    		return
    	}
    	defer file.Close()
    
    	scanner := bufio.NewScanner(file)
    	scanner.Split(bufio.ScanWords)
    
    	for scanner.Scan() {
        
    		word := scanner.Text()
    		mu.Lock()
    		wordCountMap[word]++
    		mu.Unlock()
    	}
    }
    
    func main() {
        
    	/*sync.WaitGroup 等待一组 goroutine 的执行完成。
    	它提供了三个方法：
    	Add(delta int)：WaitGroup 的计数器。
    	Done()：将 WaitGroup 的计数器减去 1。
    	Wait()：阻塞当前 goroutine，直到 WaitGroup 的计数器归零。
    	*/
    	var wg sync.WaitGroup
    	var mu sync.Mutex
    	wordCountMap := make(map[string]int)
    
    	for _, filename := range os.Args[1:] {
        
    		wg.Add(1)
    		go countWords(filename, &wg, &mu, wordCountMap)
    	}
    
    	/*主 goroutine 中调用 Add 来设置要等待的 goroutine 数量，
    	然后在每个子 goroutine 中调用 Done 来表示它已经执行完成了。
    	最后，主 goroutine 调用 Wait 来等待所有子 goroutine 执行完成。
    	*/
    	wg.Wait()
    
    	fmt.Println("word count:")
    	for word, count := range wordCountMap {
        
    		fmt.Printf("%s: %d\n", word, count)
    	}
    }

root@192:~/www/test# go run main.go test.txt
    word count:
    The: 3
    jumps: 2
    away: 1
    from: 1
    quick: 3
    over: 2
    the: 3
    lazy: 3
    dog.: 2
    dog: 1
    runs: 1
    brown: 3
    fox: 3
    again.: 1
    root@192:~/www/test#

`sync.Mutex` 是 Go 语言中的一个并发原语，用于保护共享变量的访问，避免多个 goroutine 同时访问导致数据竞争。

在 Go 语言中，多个 goroutine 可能会同时访问某个共享变量，这会导致数据竞争问题。  
为了避免这种问题，我们可以使用 sync.Mutex 来保护共享变量的访问。

sync.Mutex 提供了两个方法：

*  Lock()：锁定 Mutex，禁止其他 goroutine 访问受保护的资源。
 *  Unlock()：解锁 Mutex，允许其他 goroutine 访问受保护的资源。

当一个 goroutine 获得了 Mutex 的锁时，其他 goroutine 会被阻塞，直到该 goroutine 释放了锁。  
这样就保证了每个时刻只有一个 goroutine 能够访问受保护的资源。

在使用 sync.Mutex 时，需要注意以下几点：

1、在每次访问共享变量之前，必须先获取 Mutex 的锁。

2、在访问完成后，必须立即释放 Mutex 的锁，以允许其他 goroutine 访问共享变量。

3、不要在持有 Mutex 的锁时执行耗时的操作，因为这会导致其他 goroutine 被阻塞。

4、综上所述，sync.Mutex 是 Go 语言中的一种重要的并发原语，可以有效地保护共享变量的访问，避免数据竞争问题。