40分钟学 Go 语言高并发：Go Channel使用与实践教程

Go Channel使用与实践教程

一、Channel基础概念与类型

Channel是Go语言中的一个核心类型，是一种通讯机制，用于在不同的goroutine之间通报数据。它是Go语言实现CSP（Communicating Sequential Processes）并发模子的基础。
1.1 Channel类型和特性表

特性阐明定向性双向channel、只读channel、只写channel缓冲性无缓冲channel、有缓冲channel阻塞性阻塞式操作、非阻塞式操作（select）状态开启、关闭 Channel工作流程图

1.2 Channel根本操作

1. // 创建channel
2. ch := make(chan int)        // 无缓冲
3. ch := make(chan int, 5)     // 有缓冲
5. // 发送数据
6. ch <- value
8. // 接收数据
9. value := <-ch
10. value, ok := <-ch          // 检查channel是否关闭
12. // 关闭channel
13. close(ch)
15. // 循环读取
16. for v := range ch {
17.     // 处理数据
18. }

复制代码

二、有缓冲vs无缓冲Channel

2.1 对比表

特性无缓冲Channel有缓冲Channel容量0用户指定巨细发送操作阻塞直到有接收者缓冲区未满时不阻塞接收操作阻塞直到有发送者缓冲区非空时不阻塞同步性强同步异步实用场景需要即时相应的场景处理惩罚突发请求、削峰填谷 2.2 无缓冲Channel示例

1. package main
3. import (
4.     "fmt"
5.     "time"
6. )
8. func main() {
9.     // 创建无缓冲channel
10.     ch := make(chan int)
11.    
12.     // 发送者goroutine
13.     go func() {
14.         fmt.Println("发送数据前...")
15.         ch <- 1
16.         fmt.Println("发送数据后...")
17.     }()
18.    
19.     // 让发送者goroutine有时间先执行
20.     time.Sleep(time.Second)
21.    
22.     fmt.Println("准备接收数据...")
23.     value := <-ch
24.     fmt.Printf("接收到数据: %d\n", value)
25. }

复制代码

2.3 有缓冲Channel示例

1. package main
3. import (
4.     "fmt"
5.     "time"
6. )
8. func main() {
9.     // 创建带缓冲的channel
10.     ch := make(chan int, 2)
11.    
12.     // 发送者goroutine
13.     go func() {
14.         for i := 1; i <= 3; i++ {
15.             fmt.Printf("发送数据: %d\n", i)
16.             ch <- i
17.             fmt.Printf("成功发送: %d\n", i)
18.         }
19.     }()
20.    
21.     // 等待数据发送
22.     time.Sleep(time.Second)
23.    
24.     // 接收数据
25.     for i := 1; i <= 3; i++ {
26.         value := <-ch
27.         fmt.Printf("接收到数据: %d\n", value)
28.     }
29. }

复制代码

三、常见Channel模式

3.1 生产者-消费者模式

1. package main
3. import (
4.     "fmt"
5.     "math/rand"
6.     "time"
7. )
9. // 生产者函数
10. func producer(ch chan<- int) {
11.     for i := 0; ; i++ {
12.         // 生产数据
13.         value := rand.Intn(100)
14.         ch <- value
15.         fmt.Printf("生产者生产数据: %d\n", value)
16.         time.Sleep(time.Millisecond * 500)
17.     }
18. }
20. // 消费者函数
21. func consumer(id int, ch <-chan int) {
22.     for value := range ch {
23.         fmt.Printf("消费者%d消费数据: %d\n", id, value)
24.         time.Sleep(time.Millisecond * 800)
25.     }
26. }
28. func main() {
29.     ch := make(chan int, 5)
30.    
31.     // 启动生产者
32.     go producer(ch)
33.    
34.     // 启动多个消费者
35.     for i := 0; i < 3; i++ {
36.         go consumer(i, ch)
37.     }
38.    
39.     // 运行一段时间后退出
40.     time.Sleep(time.Second * 5)
41. }

复制代码

3.2 Pipeline模式

1. package main
3. import (
4.     "fmt"
5. )
7. // 生成数字的阶段
8. func generate(nums ...int) <-chan int {
9.     out := make(chan int)
10.     go func() {
11.         defer close(out)
12.         for _, n := range nums {
13.             out <- n
14.         }
15.     }()
16.     return out
17. }
19. // 平方计算阶段
20. func square(in <-chan int) <-chan int {
21.     out := make(chan int)
22.     go func() {
23.         defer close(out)
24.         for n := range in {
25.             out <- n * n
26.         }
27.     }()
28.     return out
29. }
31. // 过滤阶段：只保留大于100的数
32. func filter(in <-chan int) <-chan int {
33.     out := make(chan int)
34.     go func() {
35.         defer close(out)
36.         for n := range in {
37.             if n > 100 {
38.                 out <- n
39.             }
40.         }
41.     }()
42.     return out
43. }
45. func main() {
46.     // 构建pipeline
47.     numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
48.     stage1 := generate(numbers...)
49.     stage2 := square(stage1)
50.     stage3 := filter(stage2)
51.    
52.     // 打印最终结果
53.     for result := range stage3 {
54.         fmt.Printf("大于100的平方数: %d\n", result)
55.     }
56. }

复制代码

3.3 扇出扇入模式

1. package main
3. import (
4.     "fmt"
5.     "sync"
6. )
8. // 扇出：多个goroutine从一个channel读取数据
9. func split(ch <-chan int, n int) []<-chan int {
10.     outputs := make([]<-chan int, n)
11.     for i := 0; i < n; i++ {
12.         outputs[i] = processData(ch)
13.     }
14.     return outputs
15. }
17. // 处理数据的worker
18. func processData(in <-chan int) <-chan int {
19.     out := make(chan int)
20.     go func() {
21.         defer close(out)
22.         for n := range in {
23.             out <- n * n // 假设这是一个耗时的处理过程
24.         }
25.     }()
26.     return out
27. }
29. // 扇入：将多个channel合并为一个channel
30. func merge(channels ...<-chan int) <-chan int {
31.     var wg sync.WaitGroup
32.     out := make(chan int)
33.    
34.     // 为每个输入channel启动一个goroutine
35.     wg.Add(len(channels))
36.     for _, c := range channels {
37.         go func(ch <-chan int) {
38.             defer wg.Done()
39.             for n := range ch {
40.                 out <- n
41.             }
42.         }(c)
43.     }
44.    
45.     // 等待所有输入处理完成后关闭输出channel
46.     go func() {
47.         wg.Wait()
48.         close(out)
49.     }()
50.    
51.     return out
52. }
54. func main() {
55.     // 创建输入channel
56.     input := make(chan int)
57.     go func() {
58.         defer close(input)
59.         for i := 1; i <= 10; i++ {
60.             input <- i
61.         }
62.     }()
63.    
64.     // 扇出：启动3个处理goroutine
65.     channels := split(input, 3)
66.    
67.     // 扇入：合并结果
68.     results := merge(channels...)
69.    
70.     // 打印结果
71.     for result := range results {
72.         fmt.Printf("处理结果: %d\n", result)
73.     }
74. }

复制代码

四、死锁防备

4.1 常见死锁场景表

场景形貌办理方案同一goroutine中同时发送和接收无缓冲channel在同一goroutine中既要发送又要接收使用不同的goroutine处理惩罚发送和接收channel未初始化使用未make的channel使用make初始化channel循环等待多个goroutine相互等待对方的channel使用select避免永世阻塞channel泄露goroutine泄露导致channel永世无法关闭正确管理goroutine生命周期 4.2 死锁防备示例

1. package main
3. import (
4.     "fmt"
5.     "time"
6. )
8. // 使用select避免死锁
9. func safeOperation(ch chan int, value int) bool {
10.     select {
11.     case ch <- value:
12.         return true
13.     case <-time.After(time.Second):
14.         fmt.Println("操作超时")
15.         return false
16.     }
17. }
19. // 使用缓冲避免死锁
20. func bufferExample() {
21.     ch := make(chan int, 1) // 使用缓冲区
22.     ch <- 1                 // 不会阻塞
23.     fmt.Println(<-ch)       // 正常接收
24. }
26. // 正确的goroutine管理
27. func properGoroutineManagement() {
28.     done := make(chan bool)
29.     ch := make(chan int)
30.    
31.     // 启动工作goroutine
32.     go func() {
33.         defer close(ch)
34.         for i := 0; i < 5; i++ {
35.             ch <- i
36.         }
37.         done <- true
38.     }()
39.    
40.     // 接收数据
41.     go func() {
42.         for value := range ch {
43.             fmt.Printf("接收到数据: %d\n", value)
44.         }
45.     }()
46.    
47.     <-done // 等待工作完成
48. }
50. func main() {
51.     // 测试安全操作
52.     ch := make(chan int)
53.     go func() {
54.         time.Sleep(time.Second * 2)
55.         <-ch
56.     }()
57.    
58.     success := safeOperation(ch, 1)
59.     fmt.Printf("操作是否成功: %v\n", success)
60.    
61.     // 测试缓冲示例
62.     bufferExample()
63.    
64.     // 测试goroutine管理
65.     properGoroutineManagement()
66. }

复制代码

五、Channel筹划建议

Channel状态转换图

• 得当使用缓冲
  
  
  
  • 当生产和消费速率不匹配时
    
  • 需要削峰填谷时
    
  • 但不要使用过大的缓冲区
    
  
  
• 正确处理惩罚channel关闭
  
  
  
  • 由发送方负责关闭channel
    
  • 不要关闭已关闭的channel
    
  • 不要向已关闭的channel发送数据
    
  
  
• 合理使用select
  
  
  
  • 处理惩罚多个channel
    
  • 实现超时控制
    
  • 提供默认分支避免阻塞
    
  
  
• 资源管理
  
  
  
  • 确保goroutine正确退出
    
  • 防止goroutine泄露
    
  • 得当清理资源
    
  
  
• Channel性能测试的示例代码：
  

1. package channel_test
3. import (
4.     "testing"
5.     "time"
6. )
8. // 测试无缓冲channel的性能
9. func BenchmarkUnbufferedChannel(b *testing.B) {
10.     ch := make(chan int)
11.     done := make(chan bool)
13.     // 接收者goroutine
14.     go func() {
15.         for i := 0; i < b.N; i++ {
16.             <-ch
17.         }
18.         done <- true
19.     }()
21.     b.ResetTimer()
22.     // 发送数据
23.     for i := 0; i < b.N; i++ {
24.         ch <- i
25.     }
26.     <-done
27. }
29. // 测试有缓冲channel的性能
30. func BenchmarkBufferedChannel(b *testing.B) {
31.     ch := make(chan int, 1000)
32.     done := make(chan bool)
34.     // 接收者goroutine
35.     go func() {
36.         for i := 0; i < b.N; i++ {
37.             <-ch
38.         }
39.         done <- true
40.     }()
42.     b.ResetTimer()
43.     // 发送数据
44.     for i := 0; i < b.N; i++ {
45.         ch <- i
46.     }
47.     <-done
48. }
50. // 测试channel在高并发下的性能
51. func BenchmarkChannelHighConcurrency(b *testing.B) {
52.     ch := make(chan int, 1000)
53.     done := make(chan bool)
54.     concurrency := 100
56.     // 启动多个接收者
57.     for i := 0; i < concurrency; i++ {
58.         go func() {
59.             for {
60.                 select {
61.                 case <-ch:
62.                 case <-done:
63.                     return
64.                 }
65.             }
66.         }()
67.     }
69.     b.ResetTimer()
70.     // 发送数据
71.     for i := 0; i < b.N; i++ {
72.         ch <- i
73.     }
75.     close(done)
76. }
78. // 测试select语句的性能
79. func BenchmarkSelectStatement(b *testing.B) {
80.     ch1 := make(chan int)
81.     ch2 := make(chan int)
82.     done := make(chan bool)
84.     // 接收者goroutine
85.     go func() {
86.         for i := 0; i < b.N; i++ {
87.             select {
88.             case <-ch1:
89.             case <-ch2:
90.             }
91.         }
92.         done <- true
93.     }()
95.     b.ResetTimer()
96.     // 交替发送数据到两个channel
97.     for i := 0; i < b.N; i++ {
98.         if i%2 == 0 {
99.             ch1 <- i
100.         } else {
101.             ch2 <- i
102.         }
103.     }
104.     <-done
105. }
107. // 测试channel的关闭性能
108. func BenchmarkChannelClose(b *testing.B) {
109.     for i := 0; i < b.N; i++ {
110.         ch := make(chan int, 100)
111.         for j := 0; j < 100; j++ {
112.             ch <- j
113.         }
114.         close(ch)
115.         // 读取所有数据直到channel关闭
116.         for range ch {
117.         }
118.     }
119. }
121. // 测试超时控制的性能
122. func BenchmarkChannelTimeout(b *testing.B) {
123.     ch := make(chan int)
124.     timeout := time.After(time.Millisecond)
126.     b.ResetTimer()
127.     for i := 0; i < b.N; i++ {
128.         select {
129.         case ch <- i:
130.         case <-timeout:
131.         default:
132.         }
133.     }
134. }

复制代码

六、性能考虑

• 缓冲区巨细
  
  
  
  • 根据实际需求设置
    
  • 考虑内存使用
    
  • 避免过大缓冲
    
  
  
• goroutine数目
  
  
  
  • 控制并发级别
    
  • 避免过多goroutine
    
  • 考虑资源消耗
    
  
  
• channel数目
  
  
  
  • 合理筹划channel数目
    
  • 避免过多channel
    
  • 得当复用channel
    
  
  

---

怎么样本日的内容还满意吗？再次感谢观众老爷的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。末了，祝您早日实现财务自由，还请给个赞，谢谢！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。