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

🕗 发布于 2024-11-22 13:20 golang 开发语言 后端架构

Go Channel使用与实践教程

一、Channel基础概念与类型

Channel是Go语言中的一个核心类型，是一种通信机制，用于在不同的goroutine之间传递数据。它是Go语言实现CSP（Communicating Sequential Processes）并发模型的基础。

1.1 Channel类型和特性表

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

Channel工作流程图

在这里插入图片描述

1.2 Channel基本操作

// 创建channel
ch := make(chan int)        // 无缓冲
ch := make(chan int, 5)     // 有缓冲

// 发送数据
ch <- value

// 接收数据
value := <-ch
value, ok := <-ch          // 检查channel是否关闭

// 关闭channel
close(ch)

// 循环读取
for v := range ch {
    // 处理数据
}

二、有缓冲vs无缓冲Channel

2.1 对比表

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

2.2 无缓冲Channel示例

package main

import (
    "fmt"
    "time"
)

func main() {
    // 创建无缓冲channel
    ch := make(chan int)
    
    // 发送者goroutine
    go func() {
        fmt.Println("发送数据前...")
        ch <- 1
        fmt.Println("发送数据后...")
    }()
    
    // 让发送者goroutine有时间先执行
    time.Sleep(time.Second)
    
    fmt.Println("准备接收数据...")
    value := <-ch
    fmt.Printf("接收到数据: %d\n", value)
}

2.3 有缓冲Channel示例

package main

import (
    "fmt"
    "time"
)

func main() {
    // 创建带缓冲的channel
    ch := make(chan int, 2)
    
    // 发送者goroutine
    go func() {
        for i := 1; i <= 3; i++ {
            fmt.Printf("发送数据: %d\n", i)
            ch <- i
            fmt.Printf("成功发送: %d\n", i)
        }
    }()
    
    // 等待数据发送
    time.Sleep(time.Second)
    
    // 接收数据
    for i := 1; i <= 3; i++ {
        value := <-ch
        fmt.Printf("接收到数据: %d\n", value)
    }
}

三、常见Channel模式

3.1 生产者-消费者模式

package main

import (
    "fmt"
    "math/rand"
    "time"
)

// 生产者函数
func producer(ch chan<- int) {
    for i := 0; ; i++ {
        // 生产数据
        value := rand.Intn(100)
        ch <- value
        fmt.Printf("生产者生产数据: %d\n", value)
        time.Sleep(time.Millisecond * 500)
    }
}

// 消费者函数
func consumer(id int, ch <-chan int) {
    for value := range ch {
        fmt.Printf("消费者%d消费数据: %d\n", id, value)
        time.Sleep(time.Millisecond * 800)
    }
}

func main() {
    ch := make(chan int, 5)
    
    // 启动生产者
    go producer(ch)
    
    // 启动多个消费者
    for i := 0; i < 3; i++ {
        go consumer(i, ch)
    }
    
    // 运行一段时间后退出
    time.Sleep(time.Second * 5)
}

3.2 Pipeline模式

package main

import (
    "fmt"
)

// 生成数字的阶段
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// 平方计算阶段
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// 过滤阶段：只保留大于100的数
func filter(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if n > 100 {
                out <- n
            }
        }
    }()
    return out
}

func main() {
    // 构建pipeline
    numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    stage1 := generate(numbers...)
    stage2 := square(stage1)
    stage3 := filter(stage2)
    
    // 打印最终结果
    for result := range stage3 {
        fmt.Printf("大于100的平方数: %d\n", result)
    }
}

3.3 扇出扇入模式

package main

import (
    "fmt"
    "sync"
)

// 扇出：多个goroutine从一个channel读取数据
func split(ch <-chan int, n int) []<-chan int {
    outputs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        outputs[i] = processData(ch)
    }
    return outputs
}

// 处理数据的worker
func processData(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n // 假设这是一个耗时的处理过程
        }
    }()
    return out
}

// 扇入：将多个channel合并为一个channel
func merge(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)
    
    // 为每个输入channel启动一个goroutine
    wg.Add(len(channels))
    for _, c := range channels {
        go func(ch <-chan int) {
            defer wg.Done()
            for n := range ch {
                out <- n
            }
        }(c)
    }
    
    // 等待所有输入处理完成后关闭输出channel
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

func main() {
    // 创建输入channel
    input := make(chan int)
    go func() {
        defer close(input)
        for i := 1; i <= 10; i++ {
            input <- i
        }
    }()
    
    // 扇出：启动3个处理goroutine
    channels := split(input, 3)
    
    // 扇入：合并结果
    results := merge(channels...)
    
    // 打印结果
    for result := range results {
        fmt.Printf("处理结果: %d\n", result)
    }
}

四、死锁预防

4.1 常见死锁场景表

场景	描述	解决方案
同一goroutine中同时发送和接收	无缓冲channel在同一goroutine中既要发送又要接收	使用不同的goroutine处理发送和接收
channel未初始化	使用未make的channel	使用make初始化channel
循环等待	多个goroutine互相等待对方的channel	使用select避免永久阻塞
channel泄露	goroutine泄露导致channel永远无法关闭	正确管理goroutine生命周期

4.2 死锁预防示例

package main

import (
    "fmt"
    "time"
)

// 使用select避免死锁
func safeOperation(ch chan int, value int) bool {
    select {
    case ch <- value:
        return true
    case <-time.After(time.Second):
        fmt.Println("操作超时")
        return false
    }
}

// 使用缓冲避免死锁
func bufferExample() {
    ch := make(chan int, 1) // 使用缓冲区
    ch <- 1                 // 不会阻塞
    fmt.Println(<-ch)       // 正常接收
}

// 正确的goroutine管理
func properGoroutineManagement() {
    done := make(chan bool)
    ch := make(chan int)
    
    // 启动工作goroutine
    go func() {
        defer close(ch)
        for i := 0; i < 5; i++ {
            ch <- i
        }
        done <- true
    }()
    
    // 接收数据
    go func() {
        for value := range ch {
            fmt.Printf("接收到数据: %d\n", value)
        }
    }()
    
    <-done // 等待工作完成
}

func main() {
    // 测试安全操作
    ch := make(chan int)
    go func() {
        time.Sleep(time.Second * 2)
        <-ch
    }()
    
    success := safeOperation(ch, 1)
    fmt.Printf("操作是否成功: %v\n", success)
    
    // 测试缓冲示例
    bufferExample()
    
    // 测试goroutine管理
    properGoroutineManagement()
}

五、Channel设计建议

Channel状态转换图

在这里插入图片描述

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

package channel_test

import (
    "testing"
    "time"
)

// 测试无缓冲channel的性能
func BenchmarkUnbufferedChannel(b *testing.B) {
    ch := make(chan int)
    done := make(chan bool)

    // 接收者goroutine
    go func() {
        for i := 0; i < b.N; i++ {
            <-ch
        }
        done <- true
    }()

    b.ResetTimer()
    // 发送数据
    for i := 0; i < b.N; i++ {
        ch <- i
    }
    <-done
}

// 测试有缓冲channel的性能
func BenchmarkBufferedChannel(b *testing.B) {
    ch := make(chan int, 1000)
    done := make(chan bool)

    // 接收者goroutine
    go func() {
        for i := 0; i < b.N; i++ {
            <-ch
        }
        done <- true
    }()

    b.ResetTimer()
    // 发送数据
    for i := 0; i < b.N; i++ {
        ch <- i
    }
    <-done
}

// 测试channel在高并发下的性能
func BenchmarkChannelHighConcurrency(b *testing.B) {
    ch := make(chan int, 1000)
    done := make(chan bool)
    concurrency := 100

    // 启动多个接收者
    for i := 0; i < concurrency; i++ {
        go func() {
            for {
                select {
                case <-ch:
                case <-done:
                    return
                }
            }
        }()
    }

    b.ResetTimer()
    // 发送数据
    for i := 0; i < b.N; i++ {
        ch <- i
    }

    close(done)
}

// 测试select语句的性能
func BenchmarkSelectStatement(b *testing.B) {
    ch1 := make(chan int)
    ch2 := make(chan int)
    done := make(chan bool)

    // 接收者goroutine
    go func() {
        for i := 0; i < b.N; i++ {
            select {
            case <-ch1:
            case <-ch2:
            }
        }
        done <- true
    }()

    b.ResetTimer()
    // 交替发送数据到两个channel
    for i := 0; i < b.N; i++ {
        if i%2 == 0 {
            ch1 <- i
        } else {
            ch2 <- i
        }
    }
    <-done
}

// 测试channel的关闭性能
func BenchmarkChannelClose(b *testing.B) {
    for i := 0; i < b.N; i++ {
        ch := make(chan int, 100)
        for j := 0; j < 100; j++ {
            ch <- j
        }
        close(ch)
        // 读取所有数据直到channel关闭
        for range ch {
        }
    }
}

// 测试超时控制的性能
func BenchmarkChannelTimeout(b *testing.B) {
    ch := make(chan int)
    timeout := time.After(time.Millisecond)

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        select {
        case ch <- i:
        case <-timeout:
        default:
        }
    }
}

六、性能考虑

缓冲区大小
- 根据实际需求设置
- 考虑内存使用
- 避免过大缓冲
goroutine数量
- 控制并发级别
- 避免过多goroutine
- 考虑资源消耗
channel数量
- 合理设计channel数量
- 避免过多channel
- 适当复用channel

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

原文地址：https://blog.csdn.net/weixin_40780178/article/details/143930987

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Go 常量为什么只支持基本数据类型？
下一篇：设计模式：5、观察者模式（发布-订阅模式）

Windows电脑本地部署llamafile并接入Qwen大语言模型远程AI对话实战
Windows电脑本地部署llamafile并接入Qwen大语言模型远程AI对话实战
阅读更多2024-11-23
解决vm虚拟机中ctrl键粘滞问题（一直处于按下状态）
虚拟机中点击鼠标左键没有问题，单独按下键盘也没有问题，但是如果按下键盘的同时在按下鼠标左键就会卡住。而且100%稳定复现。发现问题原因：vm版本的问题，我的版本是17.5.0，升级成17.5.2 bu
阅读更多2024-11-23
第十章 JavaScript的应用
10.1.1 JavaScript简介1 简单性JavaScript 设计之初就考虑到了非专业程序员的使用，因此它的语法相对简单，容易上手。2 动态性JavaScript 是一种解释型语言，代码可以在
阅读更多2024-11-23
非root用户安装CUDA
4.下载完成后执行.run文件，需要添加两个参数，一个是--override，用于跳过某些检查或限制；另一个是--toolkitpath，它表示安装路径，必须是自己目录下面的路径，如果是根目录就会安装
阅读更多2024-11-23
Python操作neo4j库py2neo使用之py2neo 删除及事务相关操作（三）
【代码】## Python操作neo4j库py2neo使用之py2neo 删除及事务相关操作（三）
阅读更多2024-11-23
【通俗理解】Jensen不等式与变分分布q(z)在积分计算中的应用
Jensen不等式 #变分分布 #积分计算 #期望 #凸函数 #优化问题 #下界估计 #机器学习。
阅读更多2024-11-23
「Mac玩转仓颉内测版27」基础篇7 - 字符串类型详解
本篇将介绍 Cangjie 中的字符串类型，包括字符串的定义、字面量形式、插值表达、常用操作及应用场景，帮助开发者熟练掌握字符串的使用。
阅读更多2024-11-23
数据科学与SQL：组距分组分析 | 区间分布问题
绝对值分布分析也可以理解为组距分组分析。对于某个指标而言，一个记录对应的指标值的绝对值，肯定落在所有指标值的绝对值的最小值和最大值构成的区间内，根据一定的算法，在把这个区间划分为等距离的几个小区间，，
阅读更多2024-11-23
c++编译报C1004错误的原因及解决办法
可能是你的代码文件最后没有正确的结束符，或者是缺少了一些必要的头文件或语句。默认磁盘驱动器没有足够的空间用于临时文件，需要大约两倍于源文件的空间。编译器到达了源文件尾但未解析构造。计算结果为假的 #i
阅读更多2024-11-23
mysql 去重补全取出重复变量函数和存储过程
mysql 去重补全取出重复变量函数和存储过程
阅读更多2024-11-23

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

Go Channel使用与实践教程

一、Channel基础概念与类型

1.1 Channel类型和特性表

Channel工作流程图

1.2 Channel基本操作

二、有缓冲vs无缓冲Channel

2.1 对比表

2.2 无缓冲Channel示例

2.3 有缓冲Channel示例

三、常见Channel模式

3.1 生产者-消费者模式

3.2 Pipeline模式

3.3 扇出扇入模式

四、死锁预防

4.1 常见死锁场景表

4.2 死锁预防示例

五、Channel设计建议

Channel状态转换图

六、性能考虑

相关文章