Golang-使用带缓冲的Channel控制并发

导读

Channel 是Golang实现 并发编程 非常重要的组成部分, Channel 是一种内建的核心数据类型,需要使用make函数初始化,包括 无缓冲的Channel(unbuffered Channel)有缓冲的Channel(buffered Channel) 两种。 无缓冲的Channel(unbuffered Channel) 主要用于goroutine之间的同步, 有缓冲的Channel(buffered Channel) 主要用于异步通信、控制goroutine并发数量。

Unbuffered := make(chan int) // Unbuffered channel of integer type
buffered := make(chan int, 10)  // Buffered channel of integer type

场景

在我们的日常开发工作中,时常有需要控制并发的场景,如控制访问一个接口的并发,运维系统初始化机器服务的数量等。下面介绍一下如何使用 有缓冲的Channel(buffered Channel) 实现控制并发数量。

package main

import (
    "flag"
    "fmt"
    "time"
)

// Fake a long and difficult work.
func DoWork() {
    time.Sleep(5000 * time.Millisecond)
}

func main() {
    maxNbConcurrentGoroutines := flag.Int("maxNbConcurrentGoroutines", 5, "the number of goroutines that are allowed to run concurrently")
    nbJobs := flag.Int("nbJobs", 100, "the number of jobs that we need to do")
    flag.Parse()

    // Dummy channel to coordinate the number of concurrent goroutines.
    // This channel should be buffered otherwise we will be immediately blocked
    // when trying to fill it.
    concurrentGoroutines := make(chan struct{}, *maxNbConcurrentGoroutines)
    // The done channel indicates when a single goroutine has
    // finished its job.
    done := make(chan bool)
    // The waitForAllJobs channel allows the main program
    // to wait until we have indeed done all the jobs.
    waitForAllJobs := make(chan bool)

    // Collect all the jobs, and since the job is finished, we can
    // release another spot for a goroutine.
    go func() {
        for i := 0; i < *nbJobs; i++ {
            <-done
        }
        // We have collected all the jobs, the program
        // can now terminate
        waitForAllJobs <- true
    }()

    // Try to start nbJobs jobs
    for i := 1; i <= *nbJobs; i++ {
        fmt.Printf("ID: %v: waiting to launch!\n", i)
        // Try to receive from the concurrentGoroutines channel. When we have something,
        // it means we can start a new goroutine because another one finished.
        // Otherwise, it will block the execution until an execution
        // spot is available.
        concurrentGoroutines <- struct{}{}

        fmt.Printf("ID: %v: it's my turn!\n", i)
        go func(id int) {
            DoWork()
            fmt.Printf("ID: %v: all done!\n", id)
            done <- true
            <-concurrentGoroutines
        }(i)
    }

    // Wait for all jobs to finish
    <-waitForAllJobs
}

这里代码源自 Github ,我对其做了一些改动,让执行流程更容易理解、输出结果更加明显,下面解释一下这段程序:

  1. 使用 flag 包创建命令行参数,可以自己指定并发的数量 maxNbConcurrentGoroutines 和总共任务数量 nbJobs
maxNbConcurrentGoroutines := flag.Int("maxNbConcurrentGoroutines", 5, "the number of goroutines that are allowed to run concurrently")
nbJobs := flag.Int("nbJobs", 100, "the number of jobs that we need to do")
flag.Parse()

运行一个并发为10,总数为100的示例:

go run limit_concurrency.go -maxNbConcurrentGoroutines 10 -nbJobs 100
  1. 创建 concurrentGoroutines 作为控制并发的带缓冲的Channel, done 用于记录一个goroutine完成, waitForAllJobs 用于阻塞main函数,等待所有goroutine完成。
concurrentGoroutines := make(chan struct{}, *maxNbConcurrentGoroutines)
done := make(chan bool)
waitForAllJobs := make(chan bool)
  1. 创建一个收集所有goroutine运行状态的goroutine,当所有goroutine完成后向 waitForAllJobs 发送“完成信号”。
go func() {
        for i := 0; i < *nbJobs; i++ {
            <-done
        }
        waitForAllJobs <- true
    }()
  1. 这里我们叫它任务创建着,或者叫生成者也可以,首先会向 concurrentGoroutines 写入空的struct,因为 concurrentGoroutines 的buffer是10,所以这里不会阻塞,直到for循环执行10次,将buffer填满,同时的也创建了10个goroutine用于执行我们的任务,当任务执行完毕(这里都暂定执行5s),向 done 发送true通知第三步的goroutine我执行完了,接着读取 concurrentGoroutines ,“释放”一个空间,让其他goroutine可以进来继续执行,但是怎么都不会超出 buffer 的个数,等所有任务执行完, waitForAllJobs 收到了第三步gorotine发送的信号,整个程序结束,这就实现了控制并发。
for i := 1; i <= *nbJobs; i++ {
        fmt.Printf("ID: %v: waiting to launch!\n", i)
        concurrentGoroutines <- struct{}{}
        fmt.Printf("ID: %v: it's my turn!\n", i)
        go func(id int) {
            DoWork()
            fmt.Printf("ID: %v: all done!\n", id)
            done <- true
            <-concurrentGoroutines
        }(i)
    }
    <-waitForAllJobs

好了,小伙伴们,快运行下看看结果吧,Let's ready to Go :)

我来评几句
登录后评论

已发表评论数()

相关站点

热门文章