核心数据结构
源码runtime/chan.go/makechan
Channel底层是一个先进先出的环形队列（固定大小环形数组实现）
- full或empty就会阻塞
- send发送， recv接收并移除
- sendx表示最后一次插入元素的index
- recvx表示最后一次接收元素的index
- 发送、接收的操作符号都是 <-
nil通道
var c1 chan int
fmt.Printf("c1: %d, %d, %v\n", len(c1), cap(c1), c1) // c1: 0, 0, <nil>
c1 <- 111 // 阻塞，不报错。由于没有初始化容器，111塞不进去
<- c1 // 也阻塞，不报错，什么都拿不出来
chan零值是nil，即可以理解未被初始化通道这个容器。nil通道可以认为是一个只要操作就阻塞当前协程的容器。这种通道不要创建和使用，阻塞后无法解除，底层源码中写明了无法解除。
非缓冲通道
容量为0的通道，也叫同步通道。这种通道发送第一个元素时，如果没有接收操作就立即阻 塞，直到被接收。同样接收时，如果没有数据被发送就立即阻塞，直到有数据发送。
// 容量为0的非缓冲通道
c2 := make(chan int, 0)
fmt.Printf("c2: %d, %d, %v\n", len(c2), cap(c2), c2)
c3 := make(chan int) // 0可以省略不写
fmt.Printf("c3: %d, %d, %v\n", len(c3), cap(c3), c3)
缓冲通道
容量不为0的通道。如果通道已满，再往该通道发送数据的操作会被阻塞；如果通道为空，再从该通道接收数据的操作会被阻塞。
package main
import "fmt"
func main() {
c4 := make(chan int, 8) // 缓冲通道，容量为8，长度为0
fmt.Printf("c4: %d, %d, %v\n", len(c4), cap(c4), c4)
// 发送数据
c4 <- 111
c4 <- 222
fmt.Printf("c4: %d, %d, %v\n", len(c4), cap(c4), c4) // len 2
// 接收
<-c4
t := <-c4
fmt.Printf("%T %[1]v\n", t)
}
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
// 通道类型的参数传递不允许地址传递,形参处可定义单向通道类型
func product(c chan<- int, wg *sync.WaitGroup) {
fmt.Printf("%T %v\n", c, &c)
for i := 0; i < 10; i++ {
randN := rand.Intn(127)
time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
c <- randN
fmt.Printf("生产了数据: %v\n", randN)
}
// 关闭通道
close(c)
fmt.Println("关闭通道！！！")
wg.Add(-1)
}
func consume(c <-chan int, wg *sync.WaitGroup) {
for {
// 对于已关闭的通道，取数据时，如果没取到数据，则ok为false，如果能取到数据，则为true
r0, ok := <-c
if !ok {
fmt.Printf("收到信号%v,通道关闭了,0消费结束\n", ok)
break
}
time.Sleep(time.Millisecond * 500)
fmt.Printf("0消费了数据:%v --> %v\n", r0, string(byte(r0)))
}
wg.Done()
}
func consume1(c <-chan int, wg *sync.WaitGroup) {
// 用for range来遍历通道，在通道关闭前，相当于无限循环，通道关闭后，数据遍历完毕后，结束循环
for v := range c {
time.Sleep(time.Millisecond * 800)
fmt.Printf("1消费了数据:%v --> %v\n", v, string(byte(v)))
}
fmt.Println("通道关闭了, 1消费结束")
wg.Done()
}
func main() {
wg := sync.WaitGroup{}
c0 := make(chan int, 5)
fmt.Printf("%T %v\n", c0, &c0)
wg.Add(1)
go product(c0, &wg)
wg.Add(1)
go consume(c0, &wg)
wg.Add(1)
go consume1(c0, &wg)
wg.Wait()
}
生产与消费
单向通道
- <- chan type 这种定义表示只从一个channel里面拿，说明这是只读的
- chan <- type 这种定义表示只往一个channel里面写，说明这是只写的
通道关闭
使用close(ch)关闭一个通道
只有发送方才能关闭通道，一旦通道关闭，发送者不能再往其中发送数据，否则panic
通道关闭作用：告诉接收者再无新数据可以到达了
已经关闭的通道，若再次关闭则panic，因此不可重复关闭
当通道关闭时：
- t, ok := <-ch 或 t := <-ch 从通道中读取数据
- 正在阻塞等待通道中的数据的接收者，由于通道被关闭，接收者将不再阻塞，获取数据失败，ok为false，返回零值
- 接收者依然可以访问关闭的通道而不阻塞，如果通道内还有剩余数据，ok为true，接收数据；如果通道内剩余的数据被拿完了，继续接收不阻塞，ok为false，返回零值
通道遍历
1、nil通道，发送、接收、遍历都阻塞
2、缓冲的、未关闭的通道，相当于一个无限元素的通道，迭代不完，阻塞在等下一个元素到达。
3、缓冲的、关闭的通道，关闭后，通道不能再进入新的元素，那么相当于遍历有限个元素容器，遍历完就结束了。
4、非缓冲、未关闭通道，相当于一个无限元素的通道，迭代不完，阻塞在等下一个元素到达。
5、非缓冲、关闭通道，关闭后，通道不能在进入新的元素，那么相当于遍历有限个元素容器，遍历完就结束了。
PS：除nil通道外，对于未关闭通道，如同一个无限的容器，将一直迭代通道内元素，没有元素就阻塞；对于已关闭通道，将不能加入新的元素，迭代完当前通道内的元素，哪怕是0个元素，然后结束迭代
定时器
package main
import (
"fmt"
"time"
)
func main() {
t0 := time.NewTimer(time.Second * 3)
// 通道阻塞3秒后只能接收一次
fmt.Println(<-t0.C, "~~~~~")
// 类似心跳，指定时间跳一次，去chan中取， 通道每阻塞2秒就接收一次
ticker := time.NewTicker(time.Second * 2)
for v := range ticker.C {
fmt.Println(v)
}
}
通道死锁
channel满了，就阻塞写；channel空了，就阻塞读。容量为0的通道可以理解为0个元素就满了。 阻塞了当前协程之后会交出CPU，去执行其他协程，希望其他协程帮助自己解除阻塞。 main函数结束了，整个进程结束了。
如果在main协程中，执行语句阻塞时，环顾四周，如果已经没有其他子协程可以执行，只剩主协程自 己，解锁无望了，就自己把自己杀掉，报一个fatal error deadlock。
如果通道阻塞不在main协程中发生，而是发生在子协程中，子协程会继续阻塞着，也可能发生死锁。但是由于至少main协程是一个值得等待的希望，编译器不能帮你识别出死锁。如果真的无任何协程帮助该 协程解除阻塞状态，那么事实上该子协程解锁无望，已经死锁了。 死锁的危害可能会导致进程活着，但实际上某些协程未真正工作而阻塞，应该有良好的编码习惯，来减少死锁的出现。
struct{}型通道
如果一个结构体类型就是struct{}，说明该结构 体的实例没有数据成员，也就是实例内存占用为0。 这种类型数据构成的通道，非常节约内存，仅仅是为了传递一个信号标志。
package main
import (
"fmt"
"time"
)
func main() {
flag := make(chan struct{})
go func() {
t0 := time.NewTimer(time.Second * 5)
// 通道阻塞5秒后只能接收一次
fmt.Println(<-t0.C, "~~~~~", flag)
// 释放信号
flag <- struct{}{}
}()
// 类似心跳，指定时间跳一次，去chan中取， 通道每阻塞2秒就接收一次
ticker := time.NewTicker(time.Second * 2)
// 接收到信号，开始执行下文
<-flag
for v := range ticker.C {
fmt.Println(v)
}
}
View Code
通道多路复用
Go语言提供了select来监听多个channel。
// 永远阻塞
select {}
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
var ch0 = make(chan uint8, 2)
var ch1 = make(chan string, 3)
var ch2 = make(chan struct{})
var blood = 10000
var ch3 = make(chan int, 1)
func attck1(wg *sync.WaitGroup) {
defer wg.Done()
ticker := time.NewTicker(time.Millisecond * 300)
for blood >= 0 {
att := rand.Intn(99)
fmt.Printf("在%v, 步兵攻击了%d\n", <-ticker.C, att)
blood = <-ch3
blood -= att
ch3 <- blood
fmt.Printf("敌人剩余力量%d\n", blood)
}
}
func attck2(wg *sync.WaitGroup) {
defer wg.Done()
ticker := time.NewTicker(time.Millisecond * 700)
for blood >= 0 {
att := rand.Intn(300) + 100
fmt.Printf("在%v, 火箭军攻击了%d\n", <-ticker.C, att)
blood := <-ch3
blood -= att
ch3 <- blood
fmt.Printf("敌人剩余力量:%.2f%%\n", float32(blood)/float32(10000))
}
}
func prod(wg *sync.WaitGroup) {
for i := 0; i < 10; i++ {
n := rand.Intn(128)
switch n & 1 {
case 0:
ch0 <- byte(n)
case 1:
ch1 <- string(rune(n))
}
}
<-ch2
fmt.Println("一切就绪, 允许attack")
wg.Done()
}
func main() {
wg := sync.WaitGroup{}
wg.Add(1)
go prod(&wg)
for {
select { // 监听多路通道
case x0 := <-ch0:
fmt.Println("通道0就绪", x0)
case x1 := <-ch1:
fmt.Println("通道1就绪", x1)
case ch2 <- struct{}{}:
fmt.Println("请求attack")
goto END
}
}
END:
wg.Wait()
fmt.Println("attcking")
fmt.Println("You may lost!")
ch3 <- blood
wg.Add(1)
go attck1(&wg)
wg.Add(1)
go attck2(&wg)
wg.Wait()
fmt.Println("Enemy down")
}
通道多路复用
通道并发
Go语言采用并发同步模型叫做Communication Sequential Process通讯顺序进程，这是一种消息传递模型，在goroutine间传递消息， 而不是对数据进行加锁来实现同步访问。在goroutine之间使用channel来同步和传递数据。
- 多个协程之间通讯的管道
- 一端推入数据，一端拿走数据
- 同一时间，只有一个协程可以访问通道的数据
- 协调协程的执行顺序
如果多个线程都使用了同一个数据，就会出现竞争问题。因为线程的切换不会听从程序员的意志，时间 片用完就切换了。解决办法往往需要加锁，让其他线程不能对共享数据进行修改，从而保证逻辑正确。 但锁的引入严重影响并行效率。
通道适合数据流动的场景
- 如同管道一样，一级一级处理，一个协程处理完后，发送给其他协程
- 生产者、消费者模型，M:N
为了解决并发安全问题，可使用原子操作方法、或互斥锁或通道。
// 原子操作，支持的方法少 atomic.AddInt64(&count, 1) // 互斥锁 var mx sync.Mutex mx.Lock() count++ mx.Unlock()
通道的方法在通道多路复用代码中是有体现的
协程泄露
原因：
- 协程阻塞，未能如期结束，之后就会大量累积
- 协程阻塞最常见的原因都跟通道有关
- 由于每个协程都要占用内存，所以协程泄露也会导致内存泄露
因此，如果不知道你创建的协程何时能够结束，就不要使用它。否则可能协程泄露。
标签：协程,语言,fmt,chan,阻塞,go,main,通道 From： https://www.cnblogs.com/caibao666/p/17631640.html