go——通道
Posted 明王不动心
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了go——通道相关的知识，希望对你有一定的参考价值。
相比Erlang，go并未实现严格的并发安全。
允许全局变量、指针、引用类型这些非安全内存共享操作，就需要开发人员自行维护数据一致和完整性。
Go鼓励使用CSP通道，以通信来代替内存共享，实现并发安全。
作为CSP核心，通道（channel）是显式地，要求操作双方必须知道数据类型和具体通道，并不关心另一端操作者身份和数量。
可如果另一端未准备妥当，或消息未能及时处理时，会阻塞当前端。
相比起来，Actor是透明地，它不在乎数据类型及通道，只要知道接收者信箱即可。
默认就是异步方式，发送方消息是否被接收和处理并不关心。
从底层实现上来说，通道只是一个队列。
同步模式下，发送和接收双方配对，然后直接复制数据给对方。
如果配对失败，则置入等待队列，直到另一方出现后才被唤醒。
异步模式抢夺地则是数据缓冲槽。发送方要求有空槽可供写入，而接收方则要求有缓冲数据可读。
需求不符时，同样加入等待队列，直到有另一方写入数据或腾出空槽后被唤醒。
除传递消息（数据）外，通道还常被用作事件通知。
package main
import "fmt"
func main() {
done := make(chan struct{}) //消息传递通道
c := make(chan string) //数据传输通道
go func() {
s := <-c //接收消息
fmt.Println(s)
close(done) //关闭同道，作为结束通知
}()
c <- "hi" //发送消息
<-done //阻塞，直到数据或管道关闭
}
同步模式必须有配对操作的goroutine出现，否则会一直阻塞。
而异步模式在缓冲区未满或数据未读前，不会阻塞。
package main
import "fmt"
func main() {
c := make(chan int, 3) //创建带有三个缓冲槽地异步通道
c <- 1 //缓冲区未满不会阻塞
c <- 2
fmt.Println(<-c) //缓冲区尚有数据，不会阻塞
fmt.Println(<-c)
}
多数时候，异步通道有助于提升性能，减少队伍阻塞。
缓冲区大小仅是内部属性，不属于类型组成部分。
另外通道变量本身就是指针，可用相等操作符判断是否为同一对象或nil。
package main
import (
"fmt"
"unsafe"
)
func main() {
var a, b chan int = make(chan int, 3), make(chan int)
var c chan bool
fmt.Println(a == b) //槽位不同
fmt.Println(c == nil)
fmt.Printf("%p, %d\n", a, unsafe.Sizeof(a))
}
/*
false
true
0xc000080080, 8
*/
虽然可传递指针来避免数据复制，但须额外注意数据复制安全。
内置函数cap和len返回缓冲区大小和当前已缓存数量。
对于同步同步通道而言都返回0，据此可判断通道是同步还是异步。
package main
import "fmt"
func main() {
a, b := make(chan int), make(chan int, 3)
b <- 1
b <- 2
fmt.Println("a:", len(a), cap(a))
fmt.Println("b:", len(b), cap(b))
}
/*
a: 0 0 //给定槽位数量的就是异步
b: 2 3
*/
收发
除使用简单的发送和接收操作符外，还可以用ok-idom或range模式处理数据。
package main
import "fmt"
func main() {
done := make(chan struct{})
c := make(chan int)
go func() {
defer close(done)
for {
x, ok := <-c
if !ok {
return
}
fmt.Println(x)
}
// for x := range c {
// fmt.Println(x)
// }
}()
c <- 1
c <- 2
c <- 3
close(c)
<-done
}
对于循环接收数据，range模式更简洁一些。
package main
import "fmt"
func main() {
done := make(chan struct{})
c := make(chan int)
go func() {
defer close(done)
for x := range c {
fmt.Println(x)
}
}()
c <- 1
c <- 2
c <- 3
close(c)
<-done
}
及时用close函数关闭通道引发结束通知，否则可能会导致死锁。
通知可以是群体性的。也未必就是通知结束，可以是任何需要表达的事件。
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var wg sync.WaitGroup
ready := make(chan struct{})
for i := 0; i < 3; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
fmt.Println(id, ":ready")
<-ready //接收消息
fmt.Println(id, ":running...")
}(i)
}
time.Sleep(time.Second)
fmt.Println("ready? Go!")
close(ready) //关闭通道，发出消息
wg.Wait()
}
/*
0 :ready
1 :ready
2 :ready
ready? Go!
0 :running...
2 :running...
1 :running...
*/
一次性事件用close效率更好，没有多余开销。连续或多样性事件，
可传递不同数据标志实现，还可以使用sync.Cloud实现单播或广播事件。
对于closed或nil通道，发送和接收操作都有相应规则。
向已关闭通道发送数据，引发panic。
从已关闭接收数据，返回已缓冲数据或零值。
无论收发，nil通道都会阻塞。
package main
import (
"fmt"
)
func main() {
c := make(chan int, 3)
c <- 10
c <- 20
close(c)
for i := 0; i < cap(c)+1; i++ {
x, ok := <-c
fmt.Println(i, ":", ok, x)
}
}
/*
0 : true 10
1 : true 20
2 : false 0
3 : false 0
*/
注意，重复关闭或关闭nil通道都会引发panic错误。
单向
通道默认是双向的，并不区分发送和接收端。
但某些时候，我们可限制收发操作的方向类获得更严谨的操作逻辑。
尽管可用make创建单向通道，但那没有任何意义。
通常使用类型转换来获取单向通道，并分别赋予操作双方。
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(2)
c := make(chan int)
var send chan<- int = c
var recv <-chan int = c
go func() {
defer wg.Done()
for x := range recv {
fmt.Println(x)
}
}()
go func() {
defer wg.Done()
defer close(c)
for i := 0; i < 3; i++ {
send <- i
}
}()
wg.Wait()
}
/*
0
1
2
*/
/*
不能再单向通道上做逆向操作。
func main() {
c := make(chan int, 2)
var send chan<- int = c
var recv <-chan int = c
<-send
recv <- 1
}
close不能用于接收端。
func main() {
c := make(chan int, 2)
var recv <-chan int = c
close(recv)
}
无法将单向通道重新转换回去。
func main() {
var a,b clan int
a := make(chan int, 2)
var send chan<- int = a
var recv <-chan int = a
b = (chan int)(recv)
b = (chan int)(send)
}
选择
如果同时处理多个通道，可选用select语句。它会随机选择一个可用通道做收发操作。
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(2)
a, b := make(chan int), make(chan int) //创建两个通道
go func() {
defer wg.Done()
for {
var ( //定义三个变量
name string
x int
ok bool
)
select { //随机选择一个通道接收消息
case x, ok = <-a:
name = "a"
case x, ok = <-b:
name = "b"
}
if !ok { //如果任一通道关闭，则终止接收
return
}
fmt.Println(name, x)
}
}()
go func() {
defer wg.Done()
defer close(a)
defer close(b)
for i := 0; i < 10; i++ {
select {
case a <- i: //随机发送10次消息
case b <- i * 10:
}
}
}()
wg.Wait()
}
/*
a 0
b 10
b 20
a 3
a 4
a 5
a 6
b 70
a 8
b 90
*/
如果等全部通道消息处理结束，可将已完成通道设置为nil，这样它就会被阻塞，不再被select选中。
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(3)
a, b := make(chan int), make(chan int)
go func() {
defer wg.Done()
for {
select {
case x, ok := <-a:
if !ok {
a = nil
break
}
fmt.Println("a", x)
case x, ok := <-b:
if !ok {
b = nil
break
}
fmt.Println("b", x)
}
if a == nil && b == nil {
return
}
}
}()
go func() {
defer wg.Done()
defer close(a)
for i := 0; i < 3; i++ {
a <- i
}
}()
go func() {
defer wg.Done()
defer close(b)
for i := 0; i < 5; i++ {
a <- i
}
}()
wg.Wait()
}
即便是同一通道，也会随机选择case执行。
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(2)
c := make(chan int)
go func() { //接收端
defer wg.Done()
for {
var v int
var ok bool
select { //随机选择case
case v, ok = <-c:
fmt.Println("a1:", v)
case v, ok = <-c:
fmt.Println("a2:", v)
}
if !ok {
return
}
}
}()
go func() { //发送端
defer wg.Done()
defer close(c)
for i := 0; i < 10; i++ { //随机选择case
select {
case c <- i:
case c <- i * 10:
}
}
}()
wg.Wait()
}
/*
a1: 0
a2: 1
a2: 2
a2: 3
a2: 40
a1: 50
a1: 6
a2: 7
a2: 8
a2: 9
a2: 0
*/
当所有通道都不可用时，select会执行default语句。
如此可避开select阻塞，但须注意处理外层循环，以免陷入空耗。
package main
import (
"fmt"
"time"
)
func main() {
done := make(chan struct{})
c := make(chan int)
go func() {
defer close(done)
for {
select {
case x, ok := <-c:
if !ok {
return
}
fmt.Println("data:", x)
default: //避免select阻塞
}
fmt.Println(time.Now())
time.Sleep(time.Second)
}
}()
time.Sleep(time.Second * 5)
c <- 100
close(c)
<-done
}
/*
2018-12-03 06:52:57.1009398 +0800 CST m=+0.007029001
2018-12-03 06:52:58.1185419 +0800 CST m=+1.024631101
2018-12-03 06:52:59.1187182 +0800 CST m=+2.024807401
2018-12-03 06:53:00.1190807 +0800 CST m=+3.025169901
2018-12-03 06:53:01.1194511 +0800 CST m=+4.025540301
data: 100
2018-12-03 06:53:02.1198158 +0800 CST m=+5.025905001
*/
用default处理一些默认逻辑。
package main
import (
"fmt"
)
func main() {
done := make(chan struct{})
data := []chan int{
make(chan int, 3),
}
go func() {
defer close(done)
for i := 0; i < 10; i++ {
select {
case data[len(data)-1] <- i:
default:
data = append(data, make(chan int, 3))
}
}
}()
<-done
for i := 0; i < len(data); i++ {
c := data[i]
close(c)
for x := range c {
fmt.Println(x)
}
}
}
通常使用工厂方法将goroutine和通道绑定。
package main
import (
"fmt"
"sync"
)
type receiver struct {
sync.WaitGroup
data chan int
}
func newReceiver() *receiver {
r := &receiver{
data: make(chan int),
}
r.Add(1)
go func() {
defer r.Done()
for x := range r.data {
fmt.Println("recv:,", x)
}
}()
return r
}
func main() {
r := newReceiver()
r.data <- 1
r.data <- 2
close(r.data)
r.Wait()
}
/*
recv:, 1
recv:, 2
*/
鉴于通道本身就是一个并发安全的队列，可用作ID generator、Pool等用途。
package main
import (
)
type pool chan []byte
func newPool(cap int) pool {
return make(chan []byte, cap)
}
func (p pool) get() []byte {
var v []byte
select {
case v = <-p: //返回
default: //返回失败，新建
v = make([]byte, 10)
}
return v
}
func (p pool) put(b []byte) {
select {
case p <- b: //放回
default: //放回失败，放弃
}
}
用通道实现信号量。
package main
import (
"fmt"
"runtime"
"sync"
"time"
)
func main() {
runtime.GOMAXPROCS(4)
var wg sync.WaitGroup
sem := make(chan struct{}, 2) //最多允许两个并发同时执行
for i := 0; i < 5; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
time.Sleep(time.Second * 2)
fmt.Println(id, time.Now())
}(i)
}
wg.Wait()
}
/*
4 2018-12-03 07:23:18.054693 +0800 CST m=+2.004868201
0 2018-12-03 07:23:18.054693 +0800 CST m=+2.004868201
1 2018-12-03 07:23:20.0942525 +0800 CST m=+4.044427701
3 2018-12-03 07:23:20.0942525 +0800 CST m=+4.044427701
2 2018-12-03 07:23:22.0948917 +0800 CST m=+6.045066901
*/
以上是关于go——通道的主要内容，如果未能解决你的问题，请参考以下文章