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

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: %#v\n", 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: %#v\n", 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: %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具有这样的特性，我们在放数据和消费数据时可以不需要加锁。

				?

									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

... ...