脚本之家,脚本语言编程技术及教程分享平台!
分类导航

Python|VBS|Ruby|Lua|perl|VBA|Golang|PowerShell|Erlang|autoit|Dos|bat|

服务器之家 - 脚本之家 - Golang - golang 并发编程之生产者消费者详解

golang 并发编程之生产者消费者详解

2021-06-24 00:43hatlonely Golang

这篇文章主要介绍了golang 并发编程之生产者消费者详解,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧

golang 最吸引人的地方可能就是并发了,无论代码的编写上,还是性能上面,golang 都有绝对的优势

学习一个语言的并发特性,我喜欢实现一个生产者消费者模型,这个模型非常经典,适用于很多的并发场景,下面我通过这个模型,来简单介绍一下 golang 的并发编程

go 并发语法

协程 go

协程是 golang 并发的最小单元,类似于其他语言的线程,只不过线程的实现借助了操作系统的实现,每次线程的调度都是一次系统调用,需要从用户态切换到内核态,这是一项非常耗时的操作,因此一般的程序里面线程太多会导致大量的性能耗费在线程切换上。而在 golang 内部实现了这种调度,协程在这种调度下面的切换非常的轻量级,成百上千的协程跑在一个 golang 程序里面是很正常的事情

golang 为并发而生,启动一个协程的语法非常简单,使用 go 关键字即可

?
1
2
3
go func () {
    // do something
}

同步信号 sync.WaitGroup

多个协程之间可以通过 sync.WaitGroup 同步,这个类似于 Linux 里面的信号量

?
1
2
3
4
var wg sync.WaitGroup  // 申明一个信号量
wg.Add(1)   // 信号量加一
wg.Done()   // 信号量减一
wg.Wait()   // 信号量为正时阻塞,直到信号量为0时被唤醒

通道 chan

通道可以理解为一个消息队列,生产者往队列里面放,消费者从队列里面取。通道可以使用 close 关闭

?
1
2
3
4
ic := make(chan int, 10)  // 申明一个通道
ic <- 10        // 往通道里面放
i := <- ic      // 从通道里面取
close(ic)       // 关闭通道

生产者消费者实现

定义产品类

这个产品类根据具体的业务需求定义

?
1
2
3
4
type Product struct {
    name  int
    value int
}

生产者

如果 stop 标志不为 false,不断地往通道里面放 product,完成之后信号量完成

?
1
2
3
4
5
6
7
8
9
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: %#v\n", name, product)
        time.Sleep(time.Duration(200+rand.Intn(1000)) * time.Millisecond)
    }
    wg.Done()
}

消费者

不断地从通道里面取 product,然后作对应的处理,直到通道被关闭,并且 products 里面为空, for 循环才会终止,而这正是我们期望的

?
1
2
3
4
5
6
7
func consumer(wg *sync.WaitGroup, products <-chan Product, name int) {
    for product := range products {
        fmt.Printf("consumer %v consume a product: %#v\n", name, product)
        time.Sleep(time.Duration(200+rand.Intn(1000)) * time.Millisecond)
    }
    wg.Done()
}

主线程

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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中没有放入数据时,消费者端将会阻塞,等待生产者端放入数据。

代码的实现如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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: %d\n", r)
    }
    // 通知主线程,工作结束了,这一步可以省略
    msgChan <- 1
}
func testWriterAndReader(max int) {
    go writer(max)
    go reader(max)
    // writer 和reader的任务结束了,主线程会得到通知
    res := <-msgChan
    fmt.Printf("task is done: value=%d\n", res)
}
func main() {
    testWriterAndReader(100)
}

多生产者消费者模型

我们可以利用channel在某个时间点只能有一个协程能够访问其中的某一个数据,的特性来实现生产者消费者模型。由于channel具有这样的特性,我们在放数据和消费数据时可以不需要加锁。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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: %d\n", 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: %d\n", 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包的使用时再进一步分析。

以上为个人经验,希望能给大家一个参考,也希望大家多多支持服务器之家。如有错误或未考虑完全的地方,望不吝赐教。

原文链接:https://blog.csdn.net/hatlonely/article/details/7951910

延伸 · 阅读

精彩推荐
  • GolangGolang中Bit数组的实现方式

    Golang中Bit数组的实现方式

    这篇文章主要介绍了Golang中Bit数组的实现方式,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...

    天易独尊11682021-06-09
  • Golanggolang json.Marshal 特殊html字符被转义的解决方法

    golang json.Marshal 特殊html字符被转义的解决方法

    今天小编就为大家分享一篇golang json.Marshal 特殊html字符被转义的解决方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧 ...

    李浩的life12792020-05-27
  • Golanggolang 通过ssh代理连接mysql的操作

    golang 通过ssh代理连接mysql的操作

    这篇文章主要介绍了golang 通过ssh代理连接mysql的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...

    a165861639710342021-03-08
  • Golanggolang如何使用struct的tag属性的详细介绍

    golang如何使用struct的tag属性的详细介绍

    这篇文章主要介绍了golang如何使用struct的tag属性的详细介绍,从例子说起,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看...

    Go语言中文网11352020-05-21
  • Golanggo日志系统logrus显示文件和行号的操作

    go日志系统logrus显示文件和行号的操作

    这篇文章主要介绍了go日志系统logrus显示文件和行号的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...

    SmallQinYan12302021-02-02
  • GolangGolang通脉之数据类型详情

    Golang通脉之数据类型详情

    这篇文章主要介绍了Golang通脉之数据类型,在编程语言中标识符就是定义的具有某种意义的词,比如变量名、常量名、函数名等等,Go语言中标识符允许由...

    4272021-11-24
  • Golanggo语言制作端口扫描器

    go语言制作端口扫描器

    本文给大家分享的是使用go语言编写的TCP端口扫描器,可以选择IP范围,扫描的端口,以及多线程,有需要的小伙伴可以参考下。 ...

    脚本之家3642020-04-25
  • Golanggolang的httpserver优雅重启方法详解

    golang的httpserver优雅重启方法详解

    这篇文章主要给大家介绍了关于golang的httpserver优雅重启的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,...

    helight2992020-05-14