跳至主要內容

并发-channel

LiAngGo2023年12月17日并发大约 14 分钟约 2829 字

介绍

单纯地将函数并发执行是没有意义的。函数与函数间需要交换数据才能体现并发执行函数的意义。 虽然可以使用共享内存进行数据交换，但是共享内存在不同的goroutine中容易发生竞态问题。为了保证数据交换的正确性，必须使用互斥量对内存进行加锁，这种做法势必造成性能问题。 Go语言的并发模型是CSP（Communicating Sequential Processes），提倡通过通信共享内存而不是通过共享内存而实现通信。 如果说goroutine是Go程序并发的执行体，channel就是它们之间的连接。channel是可以让一个goroutine发送特定值到另一个goroutine的通信机制。 Go 语言中的通道（channel）是一种特殊的类型。通道像一个传送带或者队列，总是遵循先入先出（First In First Out）的规则，保证收发数据的顺序。每一个通道都是一个具体类型的导管，也就是声明channel的时候需要为其指定元素类型。 channel是线程安全的，多个协程操作同一管道时，不会发生资源竞争问题。

channel类型

channel是一种类型，一种引用类型。声明通道类型的格式如下：

var 变量 chan 元素类型

举几个例子：

var ch1 chan int   // 声明一个传递整型的通道     
var ch2 chan bool  // 声明一个传递布尔型的通道     
var ch3 chan []int // 声明一个传递int切片的通道

创建channel

通道是引用类型，通道类型的空值是nil。

var ch chan int
fmt.Println(ch) // <nil>

声明的通道后需要使用make函数初始化之后才能使用。管道是有类型的 创建channel的格式如下：

make(chan 元素类型, [缓冲大小])

channel的缓冲大小是可选的。 举几个例子：

ch4 := make(chan int)
ch5 := make(chan bool)
ch6 := make(chan []int)
package main

import "fmt"

func main() {
	// 1.创建可以存放3个int类型的管道
	var intChan chan int
	intChan = make(chan int, 3)
	//var intChan chan int = make(chan int, 3)
	// 2.看看intChan是什么
	fmt.Printf("intChan 的值是: %v, intChan本身的地址: %p\n", intChan, &intChan)
	// 3.向管道写入数据
	intChan <- 10
	num := 211
	intChan <- num

	//当我们给管道写入数据时，不能超过其容量

	// 4.看看长度和cap
	fmt.Printf("channel len=%v cap=%v\n", len(intChan), cap(intChan))

	// 5.从管道取出数据
	var num2 int
	num2 = <-intChan
	fmt.Printf("num2=%v\n", num2)
	fmt.Printf("channel len=%v cap=%v\n", len(intChan), cap(intChan))

	// 6.在没有使用协程的情况下，如果我们的管道数据已经全部取出，再取就会报告 deadlock
}
intChan 的值是: 0xc000022100, intChan本身的地址: 0xc00000a028
channel len=2 cap=3
num2=10
channel len=1 cap=3

channel的遍历和关闭

channel的关闭

使用内置函数close可以关闭channel，当channel关闭后，就不能再向channel写数据了，但是仍然可以从该channel读取数据。

package main

func main() {
	intChan := make(chan int, 3)
	intChan <- 100
	intChan <- 200
	close(intChan)
	//现在已经不能再写入数据到channel
	intChan <- 300
}
panic: send on closed channel

channel的遍历

import "fmt"

func main() {
	intChan := make(chan int, 3)
	intChan <- 100
	intChan <- 200
	close(intChan)
	//现在已经不能再写入数据到channel
	//intChan <- 300
	//管道关闭后，还可以读取
	for v := range intChan {
		fmt.Println(v)
	}

}

channel支持for--range 的方式进行遍历，请注意两个细节 1)在遍历时，如果channel没有关闭，则回出现 deadlock的错误 2)在遍历时，如果channel已经关闭，则会正常遍历数据，遍历完后，就会退出遍历。 遍历管道时不能使用普通的for循环

goroutine和channel结合

应用实例1

要求

开启一个writeData协程,向管道intChan中写入50个整数. 开启一个readData协程，从管道intChan中读取writeData写入的数据。 注意:writeData和readDate操作的是同一个管道 主线程需要等待writeData和readDate协程都完成工作才能退出【管道】

思路

image.png
image.png

代码

package main

import "fmt"

func writeData(intChan chan int) {
	for i := 1; i <= 50; i++ {
		intChan <- i
		fmt.Printf("writeData 写数据: %v\n", i)
	}
	close(intChan)
}

func readData(intChan chan int, exitChan chan bool) {
	for {
		v, ok := <-intChan
		if !ok {
			break
		}
		fmt.Printf("readData 读到数据: %v\n", v)
	}
	//readData 读取完数据后
	exitChan <- true
	close(exitChan)

}

func main() {
	//创建两个管道
	intChan := make(chan int, 50)
	exitChan := make(chan bool, 50)

	go writeData(intChan)
	go readData(intChan, exitChan)

	//主线程
	for true != <-exitChan {

	}
	fmt.Println("代码结束")

}

运行结果

writeData 写数据: 1
writeData 写数据: 2 
writeData 写数据: 3 
writeData 写数据: 4 
writeData 写数据: 5 
writeData 写数据: 6 
writeData 写数据: 7 
writeData 写数据: 8 
writeData 写数据: 9 
writeData 写数据: 10
writeData 写数据: 11
writeData 写数据: 12
writeData 写数据: 13
writeData 写数据: 14
writeData 写数据: 15
writeData 写数据: 16
writeData 写数据: 17
writeData 写数据: 18
writeData 写数据: 19
writeData 写数据: 20
writeData 写数据: 21
writeData 写数据: 22
writeData 写数据: 23
writeData 写数据: 24
writeData 写数据: 25
writeData 写数据: 26
writeData 写数据: 27
writeData 写数据: 28
writeData 写数据: 29
writeData 写数据: 30
writeData 写数据: 31
writeData 写数据: 32
writeData 写数据: 33
writeData 写数据: 34
writeData 写数据: 35
writeData 写数据: 36
writeData 写数据: 37
writeData 写数据: 38
writeData 写数据: 39
writeData 写数据: 40
writeData 写数据: 41
writeData 写数据: 42
writeData 写数据: 43
writeData 写数据: 44
writeData 写数据: 45
writeData 写数据: 46
writeData 写数据: 47
writeData 写数据: 48
writeData 写数据: 49
writeData 写数据: 50
readData 读到数据: 1
readData 读到数据: 2
readData 读到数据: 3
readData 读到数据: 4
readData 读到数据: 5
readData 读到数据: 6
readData 读到数据: 7
readData 读到数据: 8
readData 读到数据: 9
readData 读到数据: 10
readData 读到数据: 11
readData 读到数据: 12
readData 读到数据: 13
readData 读到数据: 14
readData 读到数据: 15
readData 读到数据: 16
readData 读到数据: 17
readData 读到数据: 18
readData 读到数据: 19
readData 读到数据: 20
readData 读到数据: 21
readData 读到数据: 22
readData 读到数据: 23
readData 读到数据: 24
readData 读到数据: 25
readData 读到数据: 26
readData 读到数据: 27
readData 读到数据: 28
readData 读到数据: 29
readData 读到数据: 30
readData 读到数据: 31
readData 读到数据: 32
readData 读到数据: 33
readData 读到数据: 34
readData 读到数据: 35
readData 读到数据: 36
readData 读到数据: 37
readData 读到数据: 38
readData 读到数据: 39
readData 读到数据: 40
readData 读到数据: 41
readData 读到数据: 42
readData 读到数据: 43
readData 读到数据: 44
readData 读到数据: 45
readData 读到数据: 46
readData 读到数据: 47
readData 读到数据: 48
readData 读到数据: 49
readData 读到数据: 50
代码结束

应用实例2-阻塞

package main

import "fmt"

func writeData(intChan chan int) {
	for i := 1; i <= 50; i++ {
		intChan <- i
		fmt.Printf("writeData 写数据: %v\n", i)
	}
	close(intChan)
}

func readData(intChan chan int, exitChan chan bool) {
	for {
		v, ok := <-intChan
		if !ok {
			break
		}
		fmt.Printf("readData 读到数据: %v\n", v)
	}
	//readData 读取完数据后
	exitChan <- true
	close(exitChan)

}

func main() {
	//创建两个管道
	intChan := make(chan int, 50)
	exitChan := make(chan bool, 50)

	go writeData(intChan)
	// go readData(intChan, exitChan)

	//主线程
	for true != <-exitChan {

	}
	fmt.Println("代码结束")

}
运行结果
writeData 写数据: 1
writeData 写数据: 2
writeData 写数据: 3
writeData 写数据: 4
writeData 写数据: 5
writeData 写数据: 6
writeData 写数据: 7
writeData 写数据: 8
writeData 写数据: 9
writeData 写数据: 10
writeData 写数据: 11
writeData 写数据: 12
writeData 写数据: 13
writeData 写数据: 14
writeData 写数据: 15
writeData 写数据: 16
writeData 写数据: 17
writeData 写数据: 18
writeData 写数据: 19
writeData 写数据: 20
writeData 写数据: 21
writeData 写数据: 22
writeData 写数据: 23
writeData 写数据: 24
writeData 写数据: 25
writeData 写数据: 26
writeData 写数据: 27
writeData 写数据: 28
writeData 写数据: 29
writeData 写数据: 30
writeData 写数据: 31
writeData 写数据: 32
writeData 写数据: 33
writeData 写数据: 34
writeData 写数据: 35
writeData 写数据: 36
writeData 写数据: 37
writeData 写数据: 38
writeData 写数据: 39
writeData 写数据: 40
writeData 写数据: 41
writeData 写数据: 42
writeData 写数据: 43
writeData 写数据: 44
writeData 写数据: 45
writeData 写数据: 46
writeData 写数据: 47
writeData 写数据: 48
writeData 写数据: 49
writeData 写数据: 50
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()
        D:/goproject/src/hello/hello1/bingfa/channeldemo/main.go:36 +0x9b
exit status 2

应用实例3

要求

统计1-200000 的数字中，哪些是素数?

思路

传统的方法，就是使用一个循环，循环的判断各个数是不是素数【ok】。 使用并发/并行的方式，将统计素数的任务分配给多个(n个)goroutine去完成，完成任务时间短。 image.png

代码

package main

import (
	"fmt"
	"time"
)

// 放入 200000个数
func putNum(intChan chan int) {
	for i := 1; i <= 2000000; i++ {
		intChan <- i
	}
	close(intChan)
}

func primeNum(intChan chan int, primeChan chan int, exitChan chan bool) {

	var flag bool
	for {
		//time.Sleep(time.Millisecond * 10) //如果要取出数据就加上
		num, ok := <-intChan
		if !ok { // intChan 取不到
			break
		}
		flag = true
		//判断是不是素数
		for i := 2; i <= num/2; i++ {
			if num%i == 0 {
				//不是素数
				flag = false
				break
			}
		}
		if flag {
			primeChan <- num
		}
	}
	fmt.Println("有一个 primeNum 协程因为取不到数据，退出")
	//这里还不能关闭 primeChan
	//向 exitChan 写入 true
	exitChan <- true
}

func main() {
	intChan := make(chan int, 50000)
	primeChan := make(chan int, 10000) // 放结果
	exitChan := make(chan bool, 4)     // 标识退出的管道

	start := time.Now().Unix()
	//开启一个协程，向 intChan 放入 1-200000个数
	go putNum(intChan)
	//开启4个协程，从 intChan 取出数据判断是否为素数
	for i := 0; i < 4; i++ {
		go primeNum(intChan, primeChan, exitChan)
	}

	//判断 exitChan 里面是不是有4个true
	go func() {
		for i := 0; i < 4; i++ {
			<-exitChan
		}
		end := time.Now().Unix()
		fmt.Println("使用协程耗时 = ", end-start)
		close(primeChan)
	}()

	for {
		_, ok := <-primeChan
		if !ok {
			break
		}
		//fmt.Printf("素数: %d\n", res)
	}
	fmt.Println("主线程退出")

}

20万用时1秒，200万用时53秒

channel注意事项

1.channel可以声明为只读，或者只写（默认情况下是双向的）

func main() {
	// 声明为只写
	var writeChan chan<- int
	writeChan = make(chan int, 3)
	writeChan <- 20
	// num := <-writeChan // cannot receive from send-only channel writeChan (variable of type chan<- int)

	// 声明为只读
	var readChan <-chan int
	num := readChan

}

应用场景 image.png 2.使用select可以解决从管道取数据的阻塞问题

func main() {
	//1.定义一个管道，10个数据int
	intChan := make(chan int, 10)
	for i := 0; i < 10; i++ {
		intChan <- i
	}

	//2.定义一个管道，5个数据string
	stringChan := make(chan string, 10)
	for i := 0; i < 5; i++ {
		stringChan <- "hello " + fmt.Sprintf("%d", i)
	}

	//传统方法遍历时，如果不关闭会阻塞而导致 deadlock
	//使用select方式解决
label:
	for {
		select {
		//注意：如果intChan一直没有关闭，也不会一直阻塞而deadlock
		case v := <-intChan:
			fmt.Printf("从intChan读取的数据%d\n", v)
		case v := <-stringChan:
			fmt.Printf("从istringChan读取的数据%s\n", v)
		default:
			fmt.Println("都取不到。。。")
			// return //推荐用return
			break label
		}

	}

}

3.goroutine中使用recover，解决协程中出现panic，导致程序崩溃问题。

func sayHello() {
	for i := 0; i < 10; i++ {
		time.Sleep(time.Second)
		fmt.Println("hello world")
	}
}

func test() {
	// 使用 defer + recover 解决
	defer func() {
		//捕获 panic
		if err := recover(); err != nil {
			fmt.Println("test()发生错误")
		}
	}()
	var myMap map[int]string
	myMap[0] = "go" //还没有make就直接用，肯定报错 panic: assignment to entry in nil map

}

func main() {
	go sayHello()
	go test()

	time.Sleep(time.Second)
	fmt.Println("main()")
}
上次编辑于: 2023/12/18 16:21:45
贡献者: LiAng