golang 中多个协程池如何优雅退出

[tim0991]

假设有三个协程池(A,B,C)， 三个分别处理不一样的业务且每个协程池中的 worker 数量不一致

当前数据流向为 A => B => C，任务在任一协程池中都有出现 err 导致该任务跳过的可能

在如下两种情况下:

• 进入 A 协程 100 个任务，任务不会继续向下拆分，最后从 C 出来最多也就是 100 个
• 进入 A 协程 100 个任务，后续每个协程池中都会继续拆分任务，比如转到 B 就是 500 个，继续转到 C 就是 1W 个

最后在主程序中，针对上述两种情况，有没有优雅的办法知道任务已经全部完成且让主程序退出 ？

[kwanzaa]

怎么听着像是 WaitGroup 该干的事儿

[tim0991]

如果用 sync.WaitGroup 只适合情况 1，且需要在 err 出现的时候考虑到如何 done 好像挺恶心的

[Trim21]

golang.org/x/sync/errgroup ？

[Sasasu]

激进：在每个任务的事件循环，每一轮循环都检测一下某个标识退出的全局变量是不是该退出了。如果没有循环那就每个阻塞操作都设超时，超时或完成时检测这个标识。
保守：关掉 A 的入口，然后当协程池工作线程 = 0 时计数 +1，计数等于协程池数量时退。

如果你要 Go 风格的话，起手一个全局 hash 表（带锁），里面 K 是每个 context 的指针。

激进：锁表，每个 context 都发 cancel，然后退出。
保守：关掉 A 的入口，轮训这个 hash 表，内容物为 0 时退出。

另外两个奇葩方案。

不想思考：关掉 A 的入口，sleep 一个所有任务超时的事件，退出。
做 CDN 的：用 dlopen 打开备用的 .so ，切工作函数。按照激进策略等旧的 .so 不再运行，释放旧 .so 然后移动新的 .so 。

[tim0991]

@Trim21 谢谢回复 这个包可以比较好的处理情况 1 中 err 的部分，但是无法处理情况 2

[tim0991]

@Sasasu 谢谢你提供的思路，但是对你说的内容有些疑问
> 然后当协程池工作线程 = 0 时计数 +1

如何知道某个协程中任务已经空了？

[Trim21]

@tim0991 #5 2 这种有拆分的应该是用 context 吧。。。

[CEBBCAT]

看不太懂，为什么不让 MQ 做消息转接呢？

[Mitt]

Context + WaitGroup 不是可以实现么，向下传递 Context 并附加 WaitGroup，每级往 WaitGroup 里添加真实的需要采集的数量并在出错的时候 defer wg.Done 并添加到 Context 中的 ErrArray，或者 errgroup，外层 wait 就好了，如果想让程序提前终止，就可以用带 cancel 的 context 从最外层一关，里面的所有任务就都会收到通知了

[tim0991]

@Mitt @Trim21 context 的只是起到超时或者从上到下关闭作用，并不能感知到任务是否完成吧

倒是全局使用一个 waitgroup 可以解决，只是全局变量到处飞略微有些蛋疼。。。

[lesismal]

难道不是

wg.Add(1)
pool.Go(func(){
defer wg.Done()
....
})

吗？

[tim0991]

@lesismal 是倒是是 就是 pool 会有多个

[lesismal]

@tim0991 你只要是整体的别落下 add done，每个 Go 是一个，不管 Go 的 func 里有没有失败跳过，只要 defer done 了就能确保 wg.wait 正常结束，所以，好像不应该有这个困扰

[securityCoding]

应该是想要 completefuture 那种？

[tim0991]

@lesismal 但是这样其实也有一个问题，就是你上面的代码示例中，协程数量是不固定的，那要是固定的协程数量怎么写比较优雅一点？

[Mitt]

@tim0991 #10 context 是上下文啊，用来把 waitgroup 传递下去的，让不管多下层都能用同一个 waitgroup 并且能通过 context 感知上层要主动关闭 比如超时，而上层也可以通过 waitgroup 来感知下层任务是否已经完成，不管你下层开多少个协程，你每一层都是知道数量的，添加进去就好了

[lesismal]

@tim0991 协程池本身就应该自带控制协程数量的属性，否则协程池还不如直接 go 。你看我上面写的也是 pool.Go

[lesismal]

@lesismal @tim0991 #17 只要你生产速度大于协程池消费速度，一样能充分利用这些数量的协程并发。最简单的实现，一个带容量的 chan，生产者往 chan 里写，多个协程去读，当前协程都忙、就被 chan 缓冲了、发送数量大于协程数量和 chan size 生产者就阻塞，这些细节看你怎么设计，姿势太多了，我这个库里就有好几种定制的，以前有些特殊的 hash 和时序需要所以没用其他三方的：
https://github.com/lesismal/nbio/tree/master/taskpool

其他第三方的也很多

[nuk]

给协程加 id，有任务就塞 map，ABC 三个 map 都空了不就表示任务干完了。。

[tim0991]

@Mitt @lesismal 谢谢你们的回复和思路