使用channel做goroutine同步

channel和goroutine是golang CSP并发模型的承载体，是goroutine同步绝对主角。channel的使用场景远比Mutex和WaitGroup多得多。

实现和WaitGroup一样的功能

func main() {
    ch := make(chan struct{}, 2)
    go func() {
        defer func() {
            ch <- struct{}{}
        }()
        time.Sleep(1e9)
        fmt.Println("after 1 second")
    }()

    go func() {
        defer func() {
            ch <- struct{}{}
        }()
        time.Sleep(2e9)
        fmt.Println("after 2 second")
    }()

    i := 0
    for _ = range ch {
        i++
        if i == 2 {
            close(ch)
        }
    }
    fmt.Println("the end")
}

生产消费模型

这是个常用的1对N的生产消费模型，常常用于消费redis队列。

package main
import (
    "os"
    "fmt"
    "os/signal"
    "syscall"
    "sync"
)

var quit bool = false
const THREAD_NUM  = 5
func main() {
    sigs := make(chan os.Signal, 1)
    //signal.Notify 注册这个给定的通道用于接收特定信号。
    signal.Notify(sigs, syscall.SIGINT, syscall.SIGUSR1, syscall.SIGUSR2)
    quitChan := make(chan bool)
    rowkeyChan := make(chan string, THREAD_NUM)
    go func() {
         <-sigs  //等待信号
        quit = true
        close(rowkeyChan)
    }()
    var wg sync.WaitGroup
    for i := 0; i < THREAD_NUM;i++ {
        wg.Add(1)
        go func(n int) {
            for {
                rowkey,ok := <- rowkeyChan
                if !ok {
                    break
                }
                //do something with rowkey
                fmt.Println(rowkey)
            }
            wg.Done()
        }(i)
    }
    go func() {
        wg.Wait()
        quitChan <- true
    }()

    for  !quit {
        //rowkey 可能来着redis的队列
        rowkey := ""
        rowkeyChan <- rowkey
    }
    <- quitChan
}

上面的代码稍微修改下很容易支持N：M的生产消费模型，这边就不再赘述。

赛道模型

在现实很多场景我们需要并发的做个任务，我们想知道他们的先后顺序，或者只想知道最快的那个。channel的特性很容易做到这个需求

package main
import (
    "fmt"
    "net/http"
)
func main() {
    ch := make(chan string)
    go func() {
        http.Head("https://www.taobao.com")
        ch <- "taobao"
    }()

    go func() {
        http.Head("https://www.jd.com")
        ch <- "jd"
    }()
    go func() {
        http.Head("https://www.vip.com")
        ch <- "vip"
    }()

    //只想知道最快的
    dm := <- ch
    fmt.Println(dm)

/* 知道它们的排名
    for dm:= range ch {
        fmt.Println(dm)
    }
    */
}

如何控制并发数

虽然golang新开一个goroutine占用的资源很小，但是无谓的goroutine开销对golang的调度性能很有影响，同时也会浪费cpu的资源。那么golang中如何控制并发数。

方式1——使用带缓冲的channel

package main
import (
    "fmt"
    "time"
)

type Task struct {
    Id int
}

func work(task *Task,limit chan struct{})  {
    time.Sleep(1e9)
    fmt.Println(task.Id)
    <- limit
}
func main() {
    ch := make(chan *Task,10)
    limit := make(chan struct{},3)
    go func() {
        for i:=0;i<10;i++ {
            ch <- &Task{i}
        }
        close(ch)
    }()

    for {
        task ,ok :=<- ch
        if ok {
            limit <- struct{}{}
            go work(task,limit)
        }
    }
}

方式2——使用master-worker模型，worker数量固定

package main

import (
    "fmt"
    "time"
)
const WorkerNum = 3
type Task struct {
    Id int
}
func work(ch chan *Task)  {
    for {//worker中循环消费任务
        task,ok :=<- ch          
        if ok {
            time.Sleep(1e9)
            fmt.Println(task.Id)
        }
    }
}
func main() {
    ch := make(chan *Task,3)
    exit := make(chan struct{})
    go func() {
        for i:=0;i<10;i++ {
            ch <- &Task{i}
        }
        close(ch)
    }()

    for i:=0;i<WorkerNum;i++{  //控制worker数量
        go work(ch)
    }
    <- exit
}

控制goroutine优雅退出

除了我们需要控制goroutine的数量之外，我们还需要控制goroutine的生命周期同样以防止不不要的资源和性能消耗。那么接下来我们介绍下控制goroutine生命周期的办法。

方法1——goroutine超时控制

package main
import (
    "fmt"
    "time"
)
func main() {
    ch := make(chan struct{})
    task := func() {
        time.Sleep(3e9)
        ch <- struct{}{}
    }
    go task()
    select {
        case <-ch: //在3s内正常完成
            fmt.Println("task finish")
        case <-time.After(3*time.Second): //超过3秒
            fmt.Println("timeout")
    }
}

$ go run main.go 
timeout

我们使用了 slecet 和 time.After 来控制goroutine超时。

方法2——使用context.Context

package main
import (
    "context"
    "fmt"
    "time"
)
func main() {
    ctx ,cancel:= context.WithCancel(context.Background())  //使用带cancel函数的context
    task := func(ctx context.Context) {
        for {
            select {
            case <-ctx.Done(): //cancel函数已经被执行了
            default:
                time.Sleep(1e9)
                fmt.Println("hello")
            }
        }
    }
    go task(ctx)
    time.Sleep(3e9) //等待3s
    cancel()          //让goroutine退出
}

$ go run main.go
hello
hello

当然我们还可以使用 context.WithTimeout的方式

package main

import (
    "context"
    "fmt"
    "time"
)
func main() {
    exit := make(chan struct{})
    ctx,_ := context.WithTimeout(context.Background(),3*time.Second) //使用带超时的context
    task := func(ctx context.Context) {
        HE:
        for {
            select {
            case <-ctx.Done(): //cancel函数已经被执行了
                break HE       //退出for循环
            default:
                time.Sleep(1e9)
                fmt.Println("hello")
            }
        }
        exit<- struct{}{}
    }
    go task(ctx)
    <-exit
}

$ go run main.go 
hello
hello
hello

这样的用法和上面的用法的区别是可以不需要手动去调用cancel函数。

总结

本小节我们介绍了使用channel同步goroutine的方法，介绍了channel的常见使用场景，已经如果控制goroutine的几种方法。这些方法在很是实际开发中都会用到。

---