go语言实现"生产者"和"消费者"的例子

学习java的多线程的时候最经典的一个例子就是生产者消费者模型的例子，最近在研究go语言协程，发现go提供的sync包中有很多和java类似的锁工具，尝试着用锁工具配合协程实现一个“消费者”和“生产者”的例子：

其实go官方文档不建议我们使用"锁"的方式来实现同步的操作，文档描述是：

“sync包提供了基本的同步基元，如互斥锁。除了Once和WaitGroup类型，大部分都是适用于低水平程序线程，高水平的同步使用channel通信更好一些。”

原文连接：https://studygolang.com/static/pkgdoc/pkg/sync.htm

使用“同步锁”的方式

package main

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

var (
        product = 0
        lock    sync.Mutex
        cond    = sync.NewCond(&lock)
)

func producer() {
        for {
                cond.L.Lock() // 先加锁
                for product > 10 {
                        fmt.Println("生产完了！")
                        cond.Wait()
                }
                fmt.Println("生产中...", product)
                product += 1
                cond.L.Unlock()
                cond.Broadcast()
        }
}

func consumer() {
        for {
                cond.L.Lock()
                for product <= 0 {
                        fmt.Println("消费完了！")
                        cond.Wait()
                }
                fmt.Println("消费中...", product)
                product -= 1
                cond.L.Unlock()
                cond.Broadcast()
        }
}

func main() {
        go producer()
        go consumer()

        time.Sleep(time.Second * 60)
        fmt.Println("主线程结束！")
}

运行：go run main.go

输出：

生产中... 0
生产中... 1
生产中... 2
生产中... 3
生产中... 4
生产中... 5
生产中... 6
生产中... 7
生产中... 8
生产中... 9
生产中... 10
生产完了！
消费中... 11
消费中... 10
消费中... 9
消费中... 8
消费中... 7
消费中... 6
消费中... 5
消费中... 4
消费中... 3
消费中... 2
消费中... 1
消费完了！
...

可以看到输出符合我们的预期。

其实官方可给我们说了，使用锁是比较低级的一种方式，因为go天然就支持协程，在协程的情况下，我们还是用同步锁其实有点浪费协程的优势，按照官方的推荐使用channel来进行协程之间的通讯，实现类似的功能：

使用channel方式实现

直接上代码：

package main

import (
        "fmt"
        "time"
)

/*
使用channel完成消费者、生产者的例子，发现使用channel会非常的方便
*/

func producer(intChan chan int) {
        for i := 0; i < cap(intChan); i++ {
                fmt.Println("生产者：", i)
                intChan <- i
        }
        // 写完后关闭掉
        close(intChan)
}

func consumer(intChan chan int, exitChan chan bool) {
        for {
                v, ok := <-intChan
                if ok {
                        fmt.Println("消费者：", v)
                } else { // 读完了
                        break
                }
                time.Sleep(time.Second)
        }
        exitChan <- true
        close(exitChan)
}

func main() {
        intChan := make(chan int, 10) // “生产者”和“消费者”之间互相通信的桥梁，这里假设生产的元素就是int类型的数字
        exitChan := make(chan bool, 1) // 退出的channel，因为仅做为一个标志所以空间为一个元素就够了
        go producer(intChan)
        go consumer(intChan, exitChan)

        // 1) for循环的等待判断
        // for {
        //      _, ok := <-exitChan
        //      if !ok {
        //              break
        //      }
        // }
        // 2) for range 阻塞，等待关闭close channel
        for ok := range exitChan {
                fmt.Println(ok)
        }
        fmt.Println("主线程结束！")

执行：go run main.go

输出：

生产者： 0
生产者： 1
生产者： 2
生产者： 3
生产者： 4
消费者： 0
生产者： 5
生产者： 6
生产者： 7
生产者： 8
生产者： 9
消费者： 1
消费者： 2
消费者： 3
消费者： 4
消费者： 5
消费者： 6
消费者： 7
消费者： 8
消费者： 9
true
主线程结束！

channel在没有被关闭的时候被遍历，此时会被当前线程阻塞，利用这个特性来实现同步的效果，更加的灵活和方便。

看到这里可能有的小伙伴会有疑问了，既然channel可以解决使用同步锁的阻塞问题，但是你使用了channel还是会阻塞啊，这不是很矛盾么？

说的没错，是的，使用channel可以方便的实现了同步锁的功能，但是我们的程序其实因为同步的关系目前仍然还是会产生阻塞，不过既然go官方文档说了使用"锁"在go语言中是低级操作，那么官方肯定提供另外一种优雅的遍历的不阻塞的方法，是的，就是这样，这里我们引入一个关键字select，这个关键字的存在就是为了解决我们的疑问的。

使用select解决阻塞

还记得我们上面代码中被注释掉的for循环么，就是准备为select的登场使用的。话不多说，上代码：

package main

import (
        "fmt"
        "time"
)

/*
使用channel完成消费者、生产者的例子，发现使用channel会非常的方便
*/

func producer(intChan chan int) {
        for i := 0; i < cap(intChan); i++ {
                fmt.Println("生产者：", i)
                intChan <- i
        }
        // 写完后关闭掉
        close(intChan)
}

func consumer(intChan chan int, exitChan chan bool) {
        for {
                v, ok := <-intChan
                if ok {
                        fmt.Println("消费者：", v)
                } else { // 读完了
                        break
                }
                time.Sleep(time.Second)
        }
        exitChan <- true
        close(exitChan)
}

func main() {
        intChan := make(chan int, 10)
        exitChan := make(chan bool, 1)
        go producer(intChan)
        go consumer(intChan, exitChan)

        // 1) for循环的等待判断
        for {
                // _, ok := <-exitChan
                // if !ok {
                //      break
                // }
                select {
                case _, ok := <-exitChan:
                        if ok {
                                fmt.Println("执行完毕！")
                                break
                        }
                default:
                        fmt.Println("读不到，执行其他的！")
                        time.Sleep(time.Second) // 此处添加Sleep才会看到效果，否则打印太多了找不到输出
                        // break // break只是跳出select循环，可配合lable跳出
                        // return
                }
        }
        // 2) for range 阻塞，等待关闭close channel
        // for ok := range exitChan {
        //      fmt.Println(ok)
        // }
        fmt.Println("主线程结束！")
}

执行：go run main.go

输出：

读不到，执行其他的！
生产者： 0
生产者： 1
生产者： 2
生产者： 3
生产者： 4
生产者： 5
生产者： 6
生产者： 7
生产者： 8
生产者： 9
消费者： 0
读不到，执行其他的！
消费者： 1
读不到，执行其他的！
消费者： 2
读不到，执行其他的！
消费者： 3
读不到，执行其他的！
消费者： 4
读不到，执行其他的！
消费者： 5
读不到，执行其他的！
消费者： 6
读不到，执行其他的！
消费者： 7
读不到，执行其他的！
消费者： 8
读不到，执行其他的！
消费者： 9
读不到，执行其他的！
执行完毕！

我们看到，当前如果没有完成的话不会阻塞，可以继续执行其他的业务逻辑，真正做到了"非阻塞"，由此可见go语言的一些特性还是灰常好用的。

参考文献：

go中文社区：https://studygolang.com/