概述
golang 最吸引人的地方可能就是并发了,无论代码的编写上,还是性能上面,golang 都有绝对的优势
学习一个语言的并发特性,我喜欢实现一个生产者消费者模型,这个模型非常经典,适用于很多的并发场景,下面我通过这个模型,来简单介绍一下 golang 的并发编程
go 并发语法
协程 go
协程是 golang 并发的最小单元,类似于其他语言的线程,只不过线程的实现借助了操作系统的实现,每次线程的调度都是一次系统调用,需要从用户态切换到内核态,这是一项非常耗时的操作,因此一般的程序里面线程太多会导致大量的性能耗费在线程切换上。而在 golang 内部实现了这种调度,协程在这种调度下面的切换非常的轻量级,成百上千的协程跑在一个 golang 程序里面是很正常的事情
golang 为并发而生,启动一个协程的语法非常简单,使用 go 关键字即可
go func () { // do something }
同步信号 sync.WaitGroup
多个协程之间可以通过 sync.WaitGroup 同步,这个类似于 Linux 里面的信号量
var wg sync.WaitGroup // 申明一个信号量 wg.Add(1) // 信号量加一 wg.Done() // 信号量减一 wg.Wait() // 信号量为正时阻塞,直到信号量为0时被唤醒
通道 chan
通道可以理解为一个消息队列,生产者往队列里面放,消费者从队列里面取。通道可以使用 close 关闭
ic := make(chan int, 10) // 申明一个通道 ic <- 10 // 往通道里面放 i := <- ic // 从通道里面取 close(ic) // 关闭通道
生产者消费者实现
定义产品类
这个产品类根据具体的业务需求定义
type Product struct { name int value int }
生产者
如果 stop 标志不为 false,不断地往通道里面放 product,完成之后信号量完成
func producer(wg *sync.WaitGroup, products chan<- Product, name int, stop *bool) { for !*stop { product := Product{name: name, value: rand.Int()} products <- product fmt.Printf("producer %v produce a product: %#vn", name, product) time.Sleep(time.Duration(200+rand.Intn(1000)) * time.Millisecond) } wg.Done() }
消费者
不断地从通道里面取 product,然后作对应的处理,直到通道被关闭,并且 products 里面为空, for 循环才会终止,而这正是我们期望的
func consumer(wg *sync.WaitGroup, products <-chan Product, name int) { for product := range products { fmt.Printf("consumer %v consume a product: %#vn", name, product) time.Sleep(time.Duration(200+rand.Intn(1000)) * time.Millisecond) } wg.Done() }
主线程
var wgp sync.WaitGroup var wgc sync.WaitGroup stop := false products := make(chan Product, 10) // 创建 5 个生产者和 5 个消费者 for i := 0; i < 5; i++ { go producer(&wgp, products, i, &stop) go consumer(&wgc, products, i) wgp.Add(1) wgc.Add(1) } time.Sleep(time.Duration(1) * time.Second) stop = true // 设置生产者终止信号 wgp.Wait() // 等待生产者退出 close(products) // 关闭通道 wgc.Wait() // 等待消费者退出
补充:Go并发编程--通过channel实现生产者消费者模型
概述
生产者消费者模型是多线程设计的经典模型,该模型被广泛的应用到各个系统的多线程/进程模型设计中。
本文介绍了Go语言中channel的特性,并通过Go语言实现了两个生产者消费者模型。
channel的一些特性
在Go中channel是非常重要的协程通信的手段,channel是双向的通道,通过channel可以实现协程间数据的传递,通过channel也可以实现协程间的同步(后面会有介绍)。
本文介绍的生产者消费者模型主要用到了channel的以下特性:任意时刻只能有一个协程能够对channel中某一个item进行访问。
单生产者单消费者模型
把生产者和消费者都放到一个无线循环中,这个和我们的服务器端的任务处理非常相似。生产者不断的向channel中放入数据,而消费者不断的从channel中取出数据,并对数据进行处理(打印)。
由于生产者的协程不会退出,所以channel的写入会永久存在,这样当channel中没有放入数据时,消费者端将会阻塞,等待生产者端放入数据。
代码的实现如下:
package main import ( "fmt" "time" ) var ch1 chan int = make(chan int) var bufChan chan int = make(chan int, 1000) var msgChan chan int = make(chan int) func sum(a int, b int) { ch1 <- a + b } // write data to channel func writer(max int) { for { for i := 0; i < max; i++ { // 简单的向channel中放入一个整数 bufChan <- i time.Sleep(1 * time.Millisecond) //控制放入的频率 } } } // read data fro m channel func reader(max int) { for { r := <-bufChan fmt.Printf("read value: %dn", r) } // 通知主线程,工作结束了,这一步可以省略 msgChan <- 1 } func testWriterAndReader(max int) { go writer(max) go reader(max) // writer 和reader的任务结束了,主线程会得到通知 res := <-msgChan fmt.Printf("task is done: value=%dn", res) } func main() { testWriterAndReader(100) }
多生产者消费者模型
我们可以利用channel在某个时间点只能有一个协程能够访问其中的某一个数据,的特性来实现生产者消费者模型。由于channel具有这样的特性,我们在放数据和消费数据时可以不需要加锁。
package main import ( "time" "fmt" "os" ) var ch1 chan int = make(chan int) var bufChan chan int = make(chan int, 1000) var msgChan chan string = make(chan string) func sum(a int, b int) { ch1 <- a + b } // write data to channel func writer(max int) { for { for i := 0; i < max; i++ { bufChan <- i fmt.Fprintf(os.Stderr, "%v write: %dn", os.Getpid(), i) time.Sleep(10 * time.Millisecond) } } } // read data fro m channel func reader(name string) { for { r := <-bufChan fmt.Printf("%s read value: %dn", name, r) } msgChan <- name } func testWriterAndReader(max int) { // 开启多个writer的goroutine,不断地向channel中写入数据 go writer(max) go writer(max) // 开启多个reader的goroutine,不断的从channel中读取数据,并处理数据 go reader("read1") go reader("read2") go reader("read3") // 获取三个reader的任务完成状态 name1 := <-msgChan name2 := <-msgChan name3 := <-msgChan fmt.Println("%s,%s,%s: All is done!!", name1, name2, name3) } func main() { testWriterAndReader(100) }
输出如下:
read3 read value: 0
80731 write: 0
80731 write: 0
read1 read value: 0
80731 write: 1
read2 read value: 1
80731 write: 1
read3 read value: 1
80731 write: 2
read2 read value: 2
80731 write: 2
... ...
总结
本文通过channel实现了经典的生产者和消费者模型,利用了channel的特性。但要注意,当消费者的速度小于生产者时,channel就有可能产生拥塞,导致占用内存增加,所以,在实际场景中需要考虑channel的缓冲区的大小。
设置了channel的大小,当生产的数据大于channel的容量时,生产者将会阻塞,这些问题都是要在实际场景中需要考虑的。
一个解决办法就是使用一个固定的数组或切片作为环形缓冲区,而非channel,通过Sync包的机制来进行同步,实现生产者消费者模型,这样可以避免由于channel满而导致消费者端阻塞。
但,对于环形缓冲区而言,可能会覆盖老的数据,同样需要考虑具体的使用场景。关于环形缓冲区的原理和实现,在分析Sync包的使用时再进一步分析。
以上为个人经验,希望能给大家一个参考,也希望大家多多支持靠谱客。如有错误或未考虑完全的地方,望不吝赐教。
最后
以上就是温暖摩托为你收集整理的golang 并发编程之生产者消费者详解的全部内容,希望文章能够帮你解决golang 并发编程之生产者消费者详解所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错,欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复