golang 并發(fā)編程之生產(chǎn)者消費(fèi)者詳解

go func () {
    // do something
}

var wg sync.WaitGroup  // 申明一個(gè)信號(hào)量
wg.Add(1)   // 信號(hào)量加一
wg.Done()   // 信號(hào)量減一
wg.Wait()   // 信號(hào)量為正時(shí)阻塞，直到信號(hào)量為0時(shí)被喚醒

ic := make(chan int, 10)  // 申明一個(gè)通道
ic - 10        // 往通道里面放
i := - ic      // 從通道里面取
close(ic)       // 關(guān)閉通道

type Product struct {
    name  int
    value int
}

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()
}

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)
// 創(chuàng)建 5 個(gè)生產(chǎn)者和 5 個(gè)消費(fèi)者
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     // 設(shè)置生產(chǎn)者終止信號(hào)
wgp.Wait()      // 等待生產(chǎn)者退出
close(products) // 關(guān)閉通道
wgc.Wait()      // 等待消費(fèi)者退出

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++ {  // 簡(jiǎn)單的向channel中放入一個(gè)整數(shù)
            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)
    }
    // 通知主線程，工作結(jié)束了，這一步可以省略
    msgChan - 1
}
func testWriterAndReader(max int) {
    go writer(max)
    go reader(max)
    // writer 和reader的任務(wù)結(jié)束了，主線程會(huì)得到通知 
    res := -msgChan
    fmt.Printf("task is done: value=%d\n", res)
}
func main() {
    testWriterAndReader(100)
}

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) {
    // 開(kāi)啟多個(gè)writer的goroutine，不斷地向channel中寫入數(shù)據(jù)
    go writer(max)
    go writer(max)
    // 開(kāi)啟多個(gè)reader的goroutine，不斷的從channel中讀取數(shù)據(jù)，并處理數(shù)據(jù)
    go reader("read1")
    go reader("read2")
    go reader("read3")
    // 獲取三個(gè)reader的任務(wù)完成狀態(tài)
    name1 := -msgChan
    name2 := -msgChan
    name3 := -msgChan
    fmt.Println("%s,%s,%s: All is done!!", name1, name2, name3)
}
func main() {
    testWriterAndReader(100)
}