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

MR_Zzzzzz_ · · 1028 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

导读

      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 :)


有疑问加站长微信联系(非本文作者)

本文来自:简书

感谢作者:MR_Zzzzzz_

查看原文:Golang-使用带缓冲的Channel控制并发

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

1028 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传