当前位置: 首页 > web >正文

go并发编程| channel入门

channel

介绍

channel 是在 Go 的并发编程中使用的,这个工具的作用之一是 goroutine 之间通信(线程通信指的是多个线程之间通过共享数据或协作机制来协调操作,通常需要借助锁来保证同步)。Go 中推荐使用 channel(不同语言有不同的并发模型,例如 Java 使用共享内存并发模型,通过锁机制和条件变量实现线程间通信)。为什么是推荐呢?因为不用 channel,也可以用共享内存的方式进行通信,但这并不推荐(例如使用 sync.Mutex、sync.Map 等进行并发控制,虽然也能实现功能,但更容易出错、调试困难,违背了 Go “通过通信来共享内存”的设计理念),这里不展开。

  • 共享内存模式 就像是多人在同一个白板上写字,需要规矩(锁)来约定谁能写。
  • channel 模式 像是每人写好纸条,通过信封(channel)传递,彼此之间不直接碰同一个白板,避免冲突。

一句话总结:没有那么复杂,channel就是一个官方写好的数据结构,用来线程通信,很安全,放心用。

它的底层已经帮你封装好了:

  • 带缓冲的 channel:用环形队列存数据;
  • 不带缓冲的 channel:发送和接收必须同步进行;
  • 内部用 锁(Mutex)和等待队列 保证并发安全。

使用

使用make初始化,eg.make(chan int),后面要跟一个类型,就是channel中放置的数据类型。

<-ch是只读管道,ch<-是只进管道,ch是双端管道

	ch := make(chan int, 2)go func() {for i := 0; i < 3; i++ {ch <- i + 1}close(ch)}()for i := range ch {// for中i就是管道中的值fmt.Println(i)}

channel具有阻塞的特点,也就是如果在取ch时无数据,当前goroutine会阻塞,如果当前ch满了还往里面放数据,当前goroutine也会阻塞在里面。

利用这个阻塞的特点,可以实现并发线程数量的控制,并发顺序的控制:

	ch := make(chan int)// 这样goroutine只有结束后,主goroutine才会继续执行go func() {for i := 0; i < 3; i++ {fmt.Println(i)}ch <- 1}()<-ch

控制并发数量:

func TestName(t *testing.T) {ch := make(chan int, 5)wg := sync.WaitGroup{}for i := 0; i < 30; i++ {wg.Add(1)go func(i int) {ch <- 1defer wg.Done()defer func() { <-ch }() // defer必须调用一个函数fmt.Println(i)time.Sleep(1 * time.Second)}(i)}wg.Wait()
}

ch可以作为全局变量也可以作为参数传导,不过网上很多都推荐作为参数传递:

func doWork(i int, ch chan int, wg *sync.WaitGroup) {ch <- 1 // 获取令牌defer func() {<-ch // 释放令牌wg.Done()}()fmt.Println(i)time.Sleep(1 * time.Second)
}func TestName(t *testing.T) {ch := make(chan int, 5)var wg sync.WaitGroupfor i := 0; i < 30; i++ {wg.Add(1)// wg是值类型,所以要取指针,ch本身就是指针,所以不用再取指针了go doWork(i, ch, &wg)}wg.Wait()
}

解释一下为什么不用传指针:Go 中的 channelchan int 这种类型)在语义上是引用类型,其底层是一个指向 hchan 结构体的指针,所以在函数间传递 channel 实际上传的是指针的副本,多个函数操作的都是同一个 channel 实例

原理

一句话总结:Go 的 channel 本质是一个线程安全的通信结构,底层是 hchan 结构体,借助环形缓冲区 + 双向队列 + 互斥锁 + 原子操作实现 协程间的通信与同步


解释:

chan T 在底层对应的是 runtime.hchan 结构体,源码在 runtime/chan.go

type hchan struct {qcount   uint           // 队列中当前元素个数dataqsiz uint           // 队列容量buf      unsafe.Pointer // 环形队列缓冲区起始地址elemsize uint16         // 元素大小closed   uint32         // 是否关闭sendx    uint           // 发送索引(环形队列位置)recvx    uint           // 接收索引(环形队列位置)recvq    waitq          // 等待接收的 goroutine 队列sendq    waitq          // 等待发送的 goroutine 队列lock     mutex          // 保证并发安全的锁
}

channel 在底层都是一个 hchan 结构体。当是带缓冲的 channel 时,它内部会维护一个环形队列用于缓存数据;当是无缓冲的 channel 时,数据只能在发送方和接收方同时存在时直接传递,不经过缓冲区。在多 goroutine 并发使用时,channel 的所有操作都由互斥锁(mutex)保护,以确保线程安全。

缓冲时的环形结构图:

buf: [v0, v1, v2, v3, v4]↑          ↑recvx      sendx

非缓冲时:

对于无缓冲 channelmake(chan T)):

  • 没有 buf
  • 发送方必须等待接收方同时准备好,才能发送成功,真正的同步通信
  • 否则发送方会被挂起,进入 sendq 等待队列。

发送操作的伪代码:

func send(ch *hchan, val T) {lock(ch.lock) // ① 加锁,保证并发安全if ch.closed {panic("send on closed channel") // ② 关闭的 channel 不允许发送}if receiver := ch.recvq.dequeue(); receiver != nil {// ③ 有接收者在等,直接将值拷贝给接收者,唤醒它receiver.value = valready(receiver)} else if ch.qcount < ch.dataqsiz {// ④ 缓冲区未满,放入缓冲区ch.buf[ch.sendx] = valch.sendx = (ch.sendx + 1) % ch.dataqsizch.qcount++} else {// ⑤ 缓冲区已满,无接收者 —— 当前 sender 入队并阻塞enqueue(ch.sendq, currentGoroutine)park() // 将当前 goroutine 挂起}unlock(ch.lock) // ⑥ 解锁
}

接收操作的伪代码:

func recv(ch *hchan) (val T, ok bool) {lock(ch.lock) // ① 加锁if ch.qcount > 0 {// ② 缓冲区有数据,从缓冲区读取val = ch.buf[ch.recvx]ch.recvx = (ch.recvx + 1) % ch.dataqsizch.qcount--unlock(ch.lock)return val, true}if sender := ch.sendq.dequeue(); sender != nil {// ③ 有发送者在等,直接从 sender 拿值,唤醒 senderval = sender.valueready(sender)unlock(ch.lock)return val, true}if ch.closed {// ④ channel 已关闭,返回零值,ok = falseunlock(ch.lock)return zeroValue(T), false}// ⑤ 没人发,缓冲区也空,当前 receiver 入队并阻塞enqueue(ch.recvq, currentGoroutine)park() // 挂起当前 goroutine,等待唤醒unlock(ch.lock)// 被唤醒时,会从其他 sender 处获取 valreturn val, true
}

小问题

问题解答
channel 是值类型还是引用类型?引用类型,传的是底层 hchan 指针
无缓冲 channel 怎么通信?发送和接收必须同时准备好(同步通信)
关闭 channel 会怎样?再次关闭会 panic;读取会读到零值且 ok==false
发送到关闭的 channel?panic
从关闭的 channel 读取?如果还有数据没读完:正常返回数据。如果已经读完了:返回该类型的零值 + ok == false
是否可以判断 channel 是否满?不行,channel 没有 len()cap() 外的并发判断能力
channel 会内存泄漏吗?如果 goroutine 被卡在 send/recv 而没人唤醒,会泄漏
http://www.xdnf.cn/news/9815.html

相关文章:

  • react库:class-variance-authority
  • GPU层次结构(Nvidia和Apple M芯片,从硬件到pytorch)
  • pyinstaller 使用 控制台闪退解决办法
  • 家庭智能监控系统实现实时监控主要依托传感器网络
  • 长安链智能合约命令解析(全集)
  • [Windows] 摸鱼小工具:隐藏软件(重制版)
  • 深入理解 Maven 循环依赖问题及其解决方案
  • 【Python Cookbook】迭代器与生成器(四)
  • 【Java Web】速通HTML
  • 电机控制选 STM32 还是 DSP?技术选型背后的现实博弈
  • day13 leetcode-hot100-24(链表3)
  • 如何利用categraf的exec插件实现对Linux主机系统用户及密码有效期进行监控及告警?
  • 序列化与反序列化
  • 【电路笔记 TMS320F28335DSP】McBSP 从源时钟得到 生成时钟 CLKG 帧同步信号 FSG
  • 【ARM】【FPGA】【硬件开发】Chapter.1 AXI4总线协议
  • 智能穿戴新标杆:SD NAND (贴片式SD卡)与 SOC 如何定义 AI 眼镜未来技术路径
  • pikachu靶场通关笔记08 XSS关卡04-DOM型XSS
  • uniapp 开发企业微信小程序时,如何在当前页面真正销毁前或者关闭小程序前调用一个api接口
  • 华为OD机试真题——Boss的收入(分销网络提成计算)(2025A卷:100分)Java/python/JavaScript/C/C++/GO最佳实现
  • Hive自定义函数案例(UDF、UDAF、UDTF)
  • kafka学习笔记(三、消费者Consumer使用教程——从指定位置消费)
  • PostgreSQL数据库配置SSL操作说明书
  • Java互联网大厂面试:从Spring Boot到Kafka的技术深度探索
  • ai工具集:AI材料星ppt生成,让你的演示更出彩
  • Codeforces Round 1025 (Div. 2)
  • springcloud openfeign 请求报错 java.net.UnknownHostException:
  • 小型语言模型:为何“小”才是“大”?
  • 【Python】3.函数与列表
  • RFID测温芯片助力新能源产业安全与能效提升
  • C++容器进阶:深入解析unordered_map与unordered_set的前世今生