争怎路由网:是一个主要分享无线路由器安装设置经验的网站,汇总WiFi常见问题的解决方法。

并发任务的固定模式(定时任务并发)

时间:2024/4/26作者:未知来源:争怎路由网人气:

虽然我们可以自己使用 chan 来实现很多并发的框架和模式,但实际应用的时候先查查 sync 包总是没错的,比如 sync/atomic 包对原子操作进行了支持,上一个章节的 *x = *x +1,其实可以直接使用 atomic.AddUint64(&x, 1) 来实现。

标准库 sync.Once 可以实现只执行一次的功能,比如你指向创建一个对象(单件)。下面这个 born 函数永远只会造同一个人:

type person struct{} var instance *person var once sync.Once func born() *person { once.Do(func() { fmt.Println("构建 instance 对象") instance = &person{} }) return instance }

我们启动 10 个 goroutine 来测试一下:

func getInstance() { wg := sync.WaitGroup{} wg.Add(10) for i := 0; i < 10; i++ { go func() { defer wg.Done() instance := born() fmt.Printf("%p\n", instance) // 0x1254790 }() } wg.Wait() }

每次获取到的 instance 的地址都是一样的,说明是同一个对象,它的实现机制如下:

func (o *Once) Do(f func()) { if atomic.LoadUint32(&o.done) == 0 { o.doSlow(f) } } func (o *Once) doSlow(f func()) { o.m.Lock() defer o.m.Unlock() if o.done == 0 { defer atomic.StoreUint32(&o.done, 1) f() } }

首先是对通过 mutex 和 atomic.StoreUint32 配合来实现,看到这里 doSlow 函数就奇怪了,你不是有 mutex 吗? 为什么还要来一发 aotmic 的函数操作呢? doSlow 函数名其实已经说明了一切。doSlow 说明这个函数很慢,因为 mutex 是一个成本很高的操作,而整数的原子读写却轻量很多,所以 atomic.LoadUint32 先判断了一下 o.done 是不是为 0,不是的话就滚蛋了,少量的竞争者最后能到达 if o.done == 0 的判断,这个提前的判断提高了性能(请注意 defer 语句执行的顺序,对 mutex 解锁 Unlock 比设置 o.done 为 1 的操作后执行。

sync/atomic 不仅能对基本的数值类型进行原子操作,atomic.Value 还能对 interface{} 数据类型进行操作。

既然系统包这么好用,为什么不把所有的应用模式封装完呢? 有时候系统包并不好用,看下面的 mutex 的例子:

func main() { var mu sync.Mutex go func(){ fmt.Println("NO.1中国人") mu.Lock() }() mu.Unlock() }

程序的意图简洁清晰明了,看起来好像没有什么问题,但是执行大概率是报错:

fatal error: sync: unlock of unlocked mutex

因为 go func 的 goroutine 和 main 的 goroutine 执行顺序并不能保证谁先谁后,mu.Unlock 的时候 mu.Lock 还未执行,所以 Unlock 直接抛出了错误。要修改它也很麻烦,要这样写:

func main() { var mu sync.Mutex mu.Lock go func(){ fmt.Println("NO.1中国人") mu.Unock() }() mu.lock() }

是不是不好理解? 第二个 mu.Lock 的时候会等着 mu.Unlock 先执行,有点绕口。但是如果使用无缓冲的通道来重构代码不但简单而且很可靠:

func main() { exit := make(chan int) go func() { fmt.Println("NO.1中国人") <-exit }() exit <- 1 }

为什么很可靠? 因为 exit 没有缓冲,不管 exit <- 1 先执行到 还是 <-exit 先执行到,都得等待同步交接,咱们谁先到都见面聊。这个例子中 <- exit 和 exit <- 1 的位置可以交换,他们的读写本身并没有意义,chan 的类型也是无意义的,用 string、struct{} 都行,但是 chan 完成了比 mutex 很易懂的代码。把这个例子扩展一下,启动指定数量的任务:

func lockBatch(taskCount int) { waitGroup := make(chan struct{}, taskCount) for i := 0; i < cap(waitGroup); i++ { go func(index int) { fmt.Printf("NO.1中国人%d\n", index) waitGroup <- struct{}{} }(i) } for i := 0; i < cap(waitGroup); i++ { <-waitGroup } }

这不就是 sync.WaitGroup 干的事情吗? 这就是框架,或者说模式。

这个模式可以启动 N 个协程,来做 N 件事情,每个协程和任务的比例是 1:1,不过太多的协程可能不利于管理和回收,能不能抽象出 N 个协程做 M 件事情的模式呢? 为了让 goroutine 可以不断的做任务,可以使用一个有缓冲的通道来保存任务,极端的情况是只有一个 goroutine 来做任务,它不断的从缓冲通道里取出任务执行,知道通道关闭为止,由于不同的 goroutine 可以安全的从通道中读取任务,因此可以随意调节 goroutine 的数目而不用修改代码。定一个 handle 函数表示做任务的工人:

var wg sync.WaitGroup func handle(tasks chan string, index int) { defer wg.Done() for { // 读取关闭的无缓冲通道,返回零值和false // 读取关闭的有缓冲通道,先把缓存数据读完后,再读取返回零值和false task, ok := <-tasks if !ok { // 任务通道关闭了,不会再有新任务了 fmt.Printf("handle %d over\n", index) return } fmt.Printf("干活儿 %s By 工人 %d\n", task, index) } }

tasks 表示任务的需求,如果只需要一个工人,直接 go handle 就行了,多个工人只需一个循环,工人下班的标志依靠 wg 来保证,干完活就能下班,而 tasks 通道用来接收任务。

func createTasks(goroutines, taskCount int) { tasks := make(chan string, taskCount) // 启动 N wg.Add(goroutines) for i := 0; i < goroutines; i++ { go handle(tasks, i) } // 设置 M for k := 0; k <= taskCount; k++ { tasks <- fmt.Sprintf("任务 %d", k) } close(tasks) wg.Wait() }

参数 goroutines 是工人的个数(协程的个数),而 taskCount 是任务的数量。需要注意的时候,在通道 close 关闭后,通道中的数据依然可以读取直到读完后再次读取事 task 是零值(这里是空字符串,因为 chan 的类型是 string),ok 为 false,也就是没有任务了,工人都可以下班了(wg.Done),而 wg.Wait 会等所有工人都下班后,才把工厂的门关上(函数返回)。如此,就完成了 N:M 的任务模型,这里 N 和 M 可以通过调用参数传递数量,而你不用修改内部的代码。


进一步延伸这个例子,就是生产者和消费者,而生产者和消费者的数量可以是任何比例。

func producer(name string, tasks chan<- string) { index := 0 for { tasks <- fmt.Sprintf("%s-task-%d", name, index) time.Sleep(50 * time.Millisecond) index++ } } func consumer(name string, tasks <-chan string) { for task := range tasks { fmt.Printf("%s-%s has been token\n", name, task) time.Sleep(50 * time.Millisecond) } }

producer 生产者是个是循环,源源不断的造东西,而 consumer 一直从 tasks 里取货,模拟一个工厂的代码:

func createFactory(stockSize, producerCount, consumerCount int) { tasks := make(chan string, stockSize) for i := 0; i <= producerCount; i++ { go producer(fmt.Sprintf("p%d", i), tasks) } for i := 0; i <= consumerCount; i++ { go consumer(fmt.Sprintf("c%d", i), tasks) } exit := make(chan os.Signal, 1) // ctrl+c 中断程序 signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM) <-exit }

stockSize 是车间大小,表示了最多能存多少货,producerCount 是生产者个数,consumerCount 是消费者个数,当生产者 pdoducer 足够多的时候,把仓库 stockSize 塞满了就阻塞了没法再生产了,除非消费者 consumer 来拿走一些;反之如果消费者足够多的时候,仓库里没货也只能等着了,除非生产者 producer 造点货出来。

为了达到平衡,如果希望生产者以稳定的效率工作,不多不少,如何控制呢? 非常简单:

func producer2(max int, tasks chan<- string, doSth func() string) { lines := make(chan int, max) for { lines <- 1 tasks <- doSth() <-lines } }

max 表示最大的生产线,每次开一个生产线,就往通道里计数一个,生产完了,再从通道取出一个,因为通道最大是 max,满了就阻塞了,随时都保证了只有 max 条生产线存在。


下面的这个例子实现的功能是,等待一个对象完成自己的工作,和前面不同的是,它把 chan 的状态作为对象的一部分:

type request struct { data []interface{} // 模拟要处理的数据 finish chan struct{} // 完成后的标志 } func newRequest(data ...interface{}) *request { return &request{data, make(chan struct{}, 1)} } func doWork(req *request) { time.Sleep(1 * time.Second) // 模拟耗时 fmt.Println(req.data) // 工作内容就是打印 close(req.finish) // req.finish <- struct{}{} 也行 } func createWork() { req := newRequest(1, "string", true) go doWork(req) for i := 0; i < 100; i++ { time.Sleep(50 * time.Millisecond) fmt.Println("继续做事,我很忙") } <-req.finish }

createWork 函数通过并发提高了处理效率,单独开了一条线去处理 request,同时它可以继续往后做事,但是它保证在函数返回前 req 对象肯定被处理了。关于 chan 和并发,还有很多有用的固定模式,希望这几个例子可以抛砖引玉。


本章节的代码 https://github.com/developdeveloper/go-demo/tree/master/16-concurrent-pattern



关键词:并发任务的固定模式(定时任务并发)




Copyright © 2012-2018 争怎路由网(http://www.zhengzen.com) .All Rights Reserved 网站地图 友情链接

免责声明:本站资源均来自互联网收集 如有侵犯到您利益的地方请及时联系管理删除,敬请见谅!

QQ:1006262270   邮箱:kfyvi376850063@126.com   手机版