Go语言编程思想6——Channel

2023-11-03

Go语言编程思想6——Channel

Channel：goroutine和goroutine之间双向的通道

一、基本语法

创建int类型的channel
c := make(chan int)
发送数据
c <- 1
接受数据
n := <-c

func chanDemo() {
	//创建int类型的channel
     c := make(chan int)
	//接收数据
	go func() {
		for {
			n := <-c
			fmt.Println(n)
		}
	}()
	//发送数据
	c <- 1
	c <- 2
}

channel发数据必须有goroutine接受，否则会发生死锁

func chanDemo() {
	cha := make(chan int)
	cha <- 1
}
/*fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:                             
main.chanDemo(...)                                   
        D:/goworkstation/src/learngo/test.go:19      
main.main()                                          
        D:/goworkstation/src/learngo/test.go:22 +0x31*/

Channel是一等公民

Channel可以作为参数或返回值

chan<- int 只能向channel发数据
<-chan int 只能从channel收数据

func createWorker(id int) chan<- int { //发数据
	c := make(chan int)
	go func(){
		for{
			fmt.Printf("Worked %d received %c\n", id, <-c)
		}
	}()
	return c
}
func chanDemo() {
	//创建10个channel分发给10个worker
	var channels [10]chan<- int
	for i := 0; i < 10; i++ {
		channels[i] = createWorker(i)
	}
	//给10个channel分发数据
	for i := 0; i < 10; i++ {
		channels[i] <- 'a' + i
	}

	for i := 0; i < 10; i++ {
		channels[i] <- 'A' + i
	}

	time.Sleep(time.Millisecond)
}

Channel缓冲区

发数据必须有人收，否则会deadlock
缓冲区大小设为3，只有在发送三个以上数据时才会出现deadlock
以下代码不会出现死锁

func bufferedChannel() {
	c := make(chan int, 3)
	c <- 'a'
	c <- 'b'
	c <- 'c'
	time.Sleep(time.Millisecond)
}

Channel的close

发送方close：
close(c) ，结束后若channel为空接收方依旧会接收到数据——(channel具体类型的零值)

接收方判断channel是否还有值的两种方法：
n, ok := <-c ，判断是否还有值，如果没有值了ok为false
for n := range c {}，若没有值会自己跳出

func worker(id int, c chan int) {
	//读完channel内的数据就退出的两种方法
	for n := range c {
		fmt.Printf("Worker %d received %c\n", id, n)
	}
	/*
	for {
		n, ok := <-c //如果没有值了ok为false
		if !ok {
			break
		}
		fmt.Printf("Worker %d received %c\n", id, n)
	}
	*/
}

func channelClose() {
	c := make(chan int)
	go worker(0, c)
	c <- 'a'
	c <- 'b'
	c <- 'c'
	c <- 'd'
	//发送方可以close
	//接收方有两种判断方法 ok，range
	close(c) //结束后依旧会接收到数据——(channel具体类型的零值)
	time.Sleep(time.Millisecond)
}

Go语言并发执行理论基础：Communication Sequential Process (CSP)

不要用共享内存实现通信，而要用通信实现共享内存

Channel阻塞

发送者角度：对于同一个通道，发送操作（协程或者函数中的），在接收者准备好之前是阻塞的。如果chan中的数据无人接收，就无法再给通道传入其他数据。因为新的输入无法在通道非空的情况下传入。所以发送操作会等待 chan 再次变为可用状态：就是通道值被接收时（可以传入变量）。
接收者角度：对于同一个通道，接收操作是阻塞的（协程或函数中的），直到发送者可用：如果通道中没有数据，接收者就阻塞了。
通过一个简单的例子来说明：

func f1(in chan int) {
	fmt.Println(<-in)
}

func main() {
	out := make(chan int)
	out <- 2
	go f1(out)
}

运行结果：fatal error: all goroutines are asleep - deadlock!

这是由于out <- 2之前不存在对out的接收，所以，对于out <- 2来说，永远是阻塞的，即一直会等下去。
　　
将out <- 2与go f1(out)互换

func f1(in chan int) {
	fmt.Println(<-in)
}

func main() {
	out := make(chan int)
	go f1(out)
	out <- 2
}

运行结果：2

out <- 2前存在对通道的读操作，所以out <- 2 是合法的。就像前文说的，发送操作在接收者准备好之前是阻塞的。

二、使用Channel等待goroutine的结束

一个死锁的例子

channel的发送和接收都是阻塞式的
发送操作在接受者准备好之前是阻塞的，这里doWork中打印完小写字母之后给done发送了一个true，此时等待接受这个true的接受程序在后面，并没有准备好所以造成阻塞，而此时又向channel中发送打印大写字母，故造成死锁。

type worker struct {
	in   chan int
	done chan bool
}
func doWork(id int, c chan int, done chan bool) {
	for {
		fmt.Printf("Worker %d received %c\n", id, <-c)
		done <- true
	}
}
func createWorker(id int) worker {
	w := worker{
		in:   make(chan int),
		done: make(chan bool),
	}
	go doWork(id, w.in, w.done)
	return w
}
//所有channel的发送的都是阻塞式的
func chanDemo() {
	var workers [10]worker
	for i := 0; i < 10; i++ {
		workers[i] = createWorker(i)
	}
	for i, worker := range workers{
		worker.in <- 'a' + i
	}
	for i, worker := range workers {
		worker.in <- 'A' + i
	}
	for _, worker := range workers {
		<-worker.done
		<-worker.done
	}
}

解决方法：
再开一个goroutine并行

func doWork(id int, c chan int, done chan bool) {
	for {
		fmt.Printf("Worker %d received %c\n", id, <-c)
		go func() { done <- true }()
	}
}

WaitGroup的使用

等待多人完成任务

方法
Add(delta int)：添加任务个数
Wait()：等待任务完成
Done()：任务完成


type worker struct {
	in   chan int
	done func()
}

func doWork(id int, w worker) {
	for n := range w.in {
		fmt.Printf("Worker %d received %c\n", id, n)
		//函数式编程，只调用done方法，具体执行什么函数由外面的createWorker来控制
		w.done()
	}
}

func createWorker(id int, wg *sync.WaitGroup) worker {
	w := worker{
		in: make(chan int),
		done: func() {
			wg.Done()
		},
	}
	go doWork(id, w)
	return w
}

//所有channel的发送的都是阻塞式的
func chanDemo() {
	var wg sync.WaitGroup

	var workers [10]worker
	for i := 0; i < 10; i++ {
		workers[i] = createWorker(i, &wg)
	}

	wg.Add(20)
	for i, worker := range workers {
		worker.in <- 'a' + i
	}

	for i, worker := range workers {
		worker.in <- 'A' + i
	}
	wg.Wait()
}

三、Select

select 是 Go 中的一个控制结构，类似于用于通信的 switch 语句。每个 case 必须是一个通信操作，要么是发送要么是接收。
select 随机执行一个可运行的 case。如果没有 case 可运行，它将阻塞，直到有 case 可运行。一个默认的子句应该总是可运行的。

每个 case 都必须是一个通信

所有 channel 表达式都会被求值

所有被发送的表达式都会被求值

如果任意某个通信可以进行，它就执行，其他被忽略。
如果有多个 case 都可以运行，Select 会随机公平地选出一个执行。其他不会执行。
否则：
如果有 default 子句，则执行该语句。
如果没有 default 子句，select 将阻塞，直到某个通信可以运行；Go 不会重新对 channel 或值进行求值。

在Select中可以使用Nil Channel，当数据还没准备好的时候可以把Channel置为nil，这样case就不会执行

select+default可以实现类似于非阻塞式的获取，如下示例输出结果为“no value”

func main() {
	c1 := make(chan int)
	select {
	case n := <-c1:
		fmt.Println(n)
	default:
		fmt.Println("no value")
	}
}

示例

package main

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

func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
			out <- i
			i++
		}
	}()
	return out
}
func worker(id int, c chan int) {
	for n := range c {
		time.Sleep(time.Second)
		fmt.Printf("Worker %d received %d\n", id, n)
	}
}
func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c
}
func main() {
	var c1, c2 chan int = generator(), generator()
	worker := createWorker(0)
	var values []int
	tm := time.After(10 * time.Second)
	tick := time.Tick(time.Second)
	for {
		var activeWorker chan<- int
		var activeValue int

		//values中存有数据，对activeWorker初始化
		if len(values) > 0 {
			activeWorker = worker
			activeValue = values[0]
		}
		select {
		case n := <-c1:
			values = append(values, n)
		case n := <-c2:
			values = append(values, n)

		//activeWorker没有值的时候为nil，此时阻塞不会执行
		case activeWorker <- activeValue:
			values = values[1:]

		
		//相邻两个请求之间超过800ms即select阻塞时间超过800ms，则输出timeout
		case <-time.After(800 * time.Millisecond):
			fmt.Println("timeout")

		//每隔一秒输出长度
		case <-tick:
			fmt.Println("queue len =", len(values))

		//总时间10s后退出
		case <-tm:
			fmt.Println("bye")
			return
		}
	}
}

time.After():是本次监听动作的超时时间，意思就说，只有在本次select 操作中会有效，再次select 又会重新开始计时

四、传统的同步机制

WaitGroup
Mutex
Cond

传统同步机制（共享内存实现通信）较少使用，一般使用channel进行通信（通信实现共享内存）

示例

package main

import (
	"fmt"
	"sync"
	"time"
)

type atomicInt struct {
	value int
	lock  sync.Mutex
}

func (a *atomicInt) increment() {
	fmt.Println("safe increment")
	func() {
		a.lock.Lock()
		defer a.lock.Unlock()
		a.value++
	}()
}

func (a *atomicInt) get() int {
	a.lock.Lock()
	defer a.lock.Unlock()
	return a.value
}

func main() {
	var a atomicInt
	a.increment()
	go func() {
		a.increment()
	}()
	time.Sleep(time.Millisecond)
	fmt.Println(a.get())
}

五、并发编程模式

生成器，可以将其抽象为服务/任务
同时等待多个服务：两种方法
- select：知道channel的具体数量
- 给每个channel开一个goroutine：不知道channel的具体数量（注意循环变量的坑，全局只有一份全局变量，在执行的时候只会向最后一个channel发送数据，可以通过拷贝一份变量或者通过函数传参来解决）

示例

package main

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

func msgGen(name string) chan string {
	c := make(chan string)
	go func() {
		i := 0
		for {
			time.Sleep(time.Duration(rand.Intn(2000)) * time.Millisecond)
			c <- fmt.Sprintf("service %s message %d", name, i)
			i++
		}
	}()
	return c
}

//不知道有多少个channel的时候用这种fanIn
func fanIn(chs ...chan string) chan string {
	c := make(chan string)
	for _, ch := range chs {
		//如果直接用ch的话，ch只有一个值，只会取最后一个channel的值送给c
		//所以要通过一个参数拷贝ch进行传递
		go func(in chan string) {
			for {
				c <- <-in
			}
		}(ch)
	}
	/*
	变量chCopy在全局有两份，通过chCopy拷贝ch
	for _, ch := range chs {
		chCopy := ch
		go func() {
			for {
				c <- <-chCopy
			}
		}()
	}
	*/
	return c
}

//明确channel个数时用select
func fanInBySelect(c1, c2 chan string) chan string {
	c := make(chan string)
	go func() {
		for {
			select {
			case m := <-c1:
				c <- m
			case m := <-c2:
				c <- m
			}
		}
	}()
	return c
}

func main() {
	m1 := msgGen("service1")
	m2 := msgGen("service2")
	m3 := msgGen("service3")
	m := fanIn(m1, m2, m3)
	for {
		fmt.Println(<-m)
	}
}

六、任务的控制

非阻塞等待
超时机制
任务中断/退出
优雅退出

示例

package main

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

func msgGen(name string, done chan struct{}) chan string {
	c := make(chan string)
	go func() {
		i := 0
		for {
			select {
			case <-time.After(time.Duration(rand.Intn(5000)) * time.Millisecond):
				c <- fmt.Sprintf("service %s message %d", name, i)
			case <-done:
				fmt.Println("cleaning up")
				time.Sleep(2 * time.Second)
				fmt.Println("cleanup done")
				done<- struct{}{}//双向通信，对方收到done才退出
				return
			}
		}
	}()
	return c
}

//非阻塞等待
//等到返回true，没等到返回false
func nonBlockingWait(c chan string) (string, bool) {
	select {
	case m := <-c:
		return m, true
	default: //一旦阻塞就进入default
		return "", false
	}
}

//超时机制
func timeoutWait(c chan string, timeout time.Duration) (string, bool) {
	select {
	case m := <-c:
		return m, true
	case <-time.After(timeout):
		return "", false
	}
}
func main() {
	done := make(chan struct{})
	m1 := msgGen("service1", done)
	for i := 0; i < 5; i++ {
		if m, ok := timeoutWait(m1, time.Second); ok {
			fmt.Println(m)
		} else {
			fmt.Println("timeout")
		}
	}
	done <- struct{}{}
	<-done
}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)