16 go语言（golang） - 并发编程select和workerpool

select

在Go语言中，select语句用于处理多个channel的操作。它类似于switch语句，但专门用于channel通信。通过使用select，可以同时等待多个channel操作，并在其中一个操作准备好时执行相应的代码块。这对于需要处理并发任务和协调goroutine之间的通信非常有用。

基本用法

多路复用：同时监听多个channel上的数据传输。
非阻塞选择：如果没有任何case可以执行，可以使用default分支来实现非阻塞行为。
超时控制：结合time.After函数，可以实现超时机制。

select {
case <-ch1:// 如果ch1成功读取到数据，则执行该case
case ch2 <- x:// 如果x成功发送到ch2，则执行该case
default:// 如果上面都没有成功，则进入default处理流程
}

具体示例

使用select语句来等待任意一个channel的数据传入，并根据哪个通道先收到消息来决定执行哪个分支。

func Test1(t *testing.T) {ch1 := make(chan string)ch2 := make(chan string)ch3 := make(chan string)go func() {time.Sleep(100 * time.Millisecond)ch1 <- "线程1"}()go func() {time.Sleep(200 * time.Millisecond)ch2 <- "线程2"}()go func() {// 只管消费数据，阻塞着一直消费for range ch3 {}}()for i := 0; i < 5; i++ {select {// 这样写会有双重读取的问题// 因为尝试从 ch1 接收数据，然后立即再次接收数据并打印。这会导致问题，因为第二次接收是在没有检查是否有数据可用的情况下进行的，这可能会阻塞程序。//case <-ch1://	value := <-ch1//	fmt.Printf("接受到来自 %s 的消息 \n", value)case value := <-ch1:fmt.Printf("接受到来自 %s 的消息 \n", value)case value := <-ch2:fmt.Printf("接受到来自 %s 的消息 \n", value)case ch3 <- "msg":time.Sleep(800 * time.Millisecond)fmt.Printf("尝试向ch3发送消息 \n")}}}

输出

尝试向ch3发送消息 
接受到来自 线程2 的消息 
尝试向ch3发送消息 
尝试向ch3发送消息 
尝试向ch3发送消息

注意：select的case不是顺序执行的，在Go语言中，select语句的行为与switch语句不同。它并不是按顺序从上到下检查每个case，而是随机选择一个可以执行的case。这种设计是为了避免优先级问题，从而使得所有可用的channel操作都有平等的机会被选中。

default

func Test2(t *testing.T) {ch1 := make(chan string)ch2 := make(chan string)go func() {time.Sleep(100 * time.Millisecond)ch1 <- "线程1"}()go func() {time.Sleep(200 * time.Millisecond)ch2 <- "线程2"}()for i := 0; i < 3; i++ {select {case value := <-ch1:fmt.Printf("接受到来自 %s 的消息 \n", value)case value := <-ch2:fmt.Printf("接受到来自 %s 的消息 \n", value)default:fmt.Println("没有数据")time.Sleep(1000 * time.Millisecond) // 没有等待的话会直接输出三次『没有数据』}}}

工作原理

随机选择
- 当有多个case都准备好时，Go会随机选择其中一个进行执行。这意味着如果有多个channel同时可以接收或发送数据，具体哪个case会被选中是不可预测的。
- default分支比较特殊，只有在没有其他case可以执行时才会被选择。因此，它并不是与其他case一起随机选择的，而是作为一种“兜底”机制。
非阻塞检查
- 如果没有任何channel操作可以立即进行，并且提供了default分支，那么default分支将被执行。
- 如果没有default分支，则select将阻塞直到某个channel操作可以进行。
- 如果所有channel操作都无法立即进行（即没有可用的数据接收或发送），且存在一个default分支，那么select将立即执行该默认分支，而不会阻塞。
公平性
- 这种随机选择机制确保了所有准备好的通道都有机会被处理，而不会因为代码中的位置而导致某些通道总是优先于其他通道。

超时控制

使用 time.After 函数可以实现对某个操作设置超时时间。如果在指定时间内没有接收到数据，可以执行超时逻辑。

func Test3(t *testing.T) {ch := make(chan string)go func() {var random = rand.Intn(2000)// 模拟随机等待0到2秒time.Sleep(time.Duration(random) * time.Millisecond)ch <- "msg"}()select {case msg := <-ch:fmt.Println("接受到消息：", msg)case <-time.After(1 * time.Second):fmt.Println("Timeout!")}
}

关闭通道检测

当一个通道被关闭并且所有的数据都被读取完毕后，再次读取会立即返回零值，这可以通过 select 来检测。

func Test4(t *testing.T) {ch := make(chan string)go func() {ch <- "msg"}()select {case msg, ok := <-ch:if ok {fmt.Println("接受到消息：", msg)} else {fmt.Println("channel 已经关闭")}}close(ch)select {case msg, ok := <-ch:if ok {fmt.Println("接受到消息：", msg)} else {fmt.Println("channel 已经关闭")}}}

workerpool

在Golang中，Worker Pool是一种并发编程模式，用于限制同时运行的goroutine数量，以控制资源使用和提高程序的性能。通过使用Worker Pool，可以有效地管理任务执行，避免因过多的goroutine导致系统资源耗尽。

基本原理

任务队列：一个用于存储待处理任务的通道。
Workers：一组固定数量的goroutine，从任务队列中获取任务并执行。
结果收集：通常会有一个结果通道，用于收集每个worker完成后的结果。

工作流程

主线程将所有待处理的任务放入到任务队列中。
一组预先启动好的worker从该队列中获取任务进行处理。
每个worker在完成其当前工作后，会继续从队列中获取下一个可用的工作，直到所有工作都被完成。

与线程池的区别

Worker Pool与线程池在概念上非常相似。两者都是用于管理并发任务执行的设计模式，旨在限制同时运行的工作单元（goroutines或线程）的数量，以有效利用系统资源并提高性能。

相似

资源管理：都用于限制并发执行单元（如goroutine或线程）的数量，从而控制对系统资源（如CPU、内存等）的使用。
复用工作单元：通过复用已有的工作单元来减少创建和销毁它们所带来的开销，提高效率。
异步执行：允许提交大量任务，并由池中的工作单元异步地完成这些任务。

不同

实现机制：
- 在Golang中，Worker Pool通常基于goroutines实现，而不是操作系统级别的线程。这使得Golang中的Worker Pool更加轻量级，因为goroutine比传统线程更小且启动速度更快。
- 传统语言中的线程池通常直接基于操作系统提供的线程模型，这可能会导致较高的上下文切换开销和内存消耗。
调度方式：
- Golang有自己的调度器来管理goroutines，它可以自动将数千个甚至更多个goroutine映射到少量OS线程上运行。
- 传统语言中的线程池依赖于操作系统调度器来管理和分配CPU时间片给各个线程。

实现

Golang标准库中没有内置的专门用于实现Worker Pool的包或功能。不过，Golang提供了强大的goroutine和channel机制，使得实现自定义的Worker Pool变得相对简单。

Worker Pool（工作池）的实现思路主要围绕如何有效管理和调度一组有限的工作者（goroutine）来执行任务。

1、定义 Worker Pool 结构

首先，定义一个 WorkerPool 结构，它包含以下元素：

最大工作者数（maxWorkers）：控制同时运行的 goroutine 的最大数量。
任务队列（taskQueue）：用于存储待处理任务，通常使用channel来实现。
停止信号（stopSignal）：一个通道，用于发送停止信号给所有工作者，让它们停止执行。
同步机制：如互斥锁（sync.Mutex）或 WaitGroup（sync.WaitGroup），用于同步和等待所有工作者完成。

// WorkerPool
// 池
type WorkerPool struct {maxWorkerNums inttaskQueue     chan TaskstopSignal    chan int // 停止信号，接受到数据时时停止waitGroup     sync.WaitGroup// 保证停止后不能再提交任务isStop boolmu     sync.Mutex
}

2、初始化 Worker Pool

实现一个 New 函数来初始化 WorkerPool，设置初始状态，并启动一定数量的工作者。

根据 maxWorkers 初始化工作者队列。
创建 taskQueue

// NewWorkerPool
/* 初始化池*/
func NewWorkerPool(maxWorkerNums int) WorkerPool {return WorkerPool{maxWorkerNums,make(chan Task),make(chan int, maxWorkerNums),sync.WaitGroup{},false,sync.Mutex{},}
}

3、提交任务

实现一个 Submit 方法，用于提交任务到 Worker Pool。

func (p *WorkerPool) submit(task Task) {p.mu.Lock()defer p.mu.Unlock()// 前提是队列还没有关闭的情况下，提交任务if !p.isStop {p.taskQueue <- task}
}

4、启动池

工作者运行在一个无限循环中，不断从 workerQueue 中取出任务并执行。

使用 for 循环和 select 语句监听 workerQueue 中的任务。
执行任务，然后等待下一个任务。
如果接收到任务队列关闭或 stopSignal，工作者退出循环。

// worker执行任务
func (p *WorkerPool) worker(num int) {defer p.waitGroup.Done()for {select {case <-p.stopSignal:// 接受到停止信号fmt.Printf("【%d号】接受到停止讯号，结束！\n", num)returncase task, ok := <-p.taskQueue:if ok {// 具体的业务逻辑process(task, num)} else {// 任务队列被关闭了，表示没有任务了fmt.Printf("【%d号】完成！\n", num)return}default:// 暂时没有任务，睡眠10毫秒fmt.Println("！空闲！没有任务")time.Sleep(time.Millisecond * 100)}}
}

5、停止

实现一个 Stop 方法，用于优雅地关闭 Worker Pool。

发送停止信号给所有工作者，让它们结束循环。
使用 WaitGroup 等待所有工作者完成当前任务。
关闭任务队列和工作者队列。

func (p *WorkerPool) stop() {p.mu.Lock()defer p.mu.Unlock()// 发送停止讯号for i := 0; i < p.maxWorkerNums; i++ {p.stopSignal <- 1}if !p.isStop { // 确保只关闭一次。p.isStop = trueclose(p.taskQueue)}p.waitGroup.Wait()close(p.stopSignal)
}

6、具体任务和业务逻辑

// Task 具体需要执行单元任务
type Task interface{}func process(t Task, num int) {fmt.Printf("...【%d号】正在处理任务：%v\n", num, t)// 模拟耗时操作time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)fmt.Printf("✓完成任务：%v\n", t)
}

7、启动测试程序

package mainimport ("fmt""math/rand""sync""time"
)func main() {// 初始化一个工作者池pool := NewWorkerPool(5)// 初始化并开始工作pool.start()// 模拟发送1000个任务wg := sync.WaitGroup{}wg.Add(1)go func() {for i := 0; i < 1000; i++ {pool.submit(fmt.Sprintf("任务%d", i))}}()wg.Done()wg.Wait()// 模拟程序运行中time.Sleep(2 * time.Second)// 强制停止程序pool.stop()fmt.Println("main结束")
}

输出：

！空闲！没有任务
！空闲！没有任务
！空闲！没有任务
！空闲！没有任务
！空闲！没有任务
...【1号】正在处理任务：任务0
！空闲！没有任务
！空闲！没有任务
！空闲！没有任务
！空闲！没有任务
...【3号】正在处理任务：任务1
...【0号】正在处理任务：任务2
！空闲！没有任务
...【4号】正在处理任务：任务3
✓完成任务：任务3
...【4号】正在处理任务：任务4
...【2号】正在处理任务：任务5
✓完成任务：任务4
...【4号】正在处理任务：任务6
✓完成任务：任务6
...【4号】正在处理任务：任务7
✓完成任务：任务2
...【0号】正在处理任务：任务8
✓完成任务：任务5
...【2号】正在处理任务：任务9
✓完成任务：任务0
...【1号】正在处理任务：任务10
✓完成任务：任务7
...【4号】正在处理任务：任务11
✓完成任务：任务1
...【3号】正在处理任务：任务12
✓完成任务：任务8
...【0号】正在处理任务：任务13
✓完成任务：任务11
...【4号】正在处理任务：任务14
✓完成任务：任务10
...【1号】正在处理任务：任务15
✓完成任务：任务12
...【3号】正在处理任务：任务16
✓完成任务：任务13
...【0号】正在处理任务：任务17
✓完成任务：任务14
...【4号】正在处理任务：任务18
✓完成任务：任务17
...【0号】正在处理任务：任务19
✓完成任务：任务9
...【2号】正在处理任务：任务20
✓完成任务：任务15
...【1号】正在处理任务：任务21
✓完成任务：任务18
...【4号】正在处理任务：任务22
✓完成任务：任务21
...【1号】正在处理任务：任务23
✓完成任务：任务22
...【4号】正在处理任务：任务24
✓完成任务：任务24
...【4号】正在处理任务：任务25
✓完成任务：任务16
...【3号】正在处理任务：任务26
✓完成任务：任务23
...【1号】正在处理任务：任务27
✓完成任务：任务20
...【2号】正在处理任务：任务28
✓完成任务：任务26
【3号】完成！
✓完成任务：任务19
【0号】完成！
✓完成任务：任务27
【1号】完成！
✓完成任务：任务25
【4号】接受到停止讯号，结束！
✓完成任务：任务28
【2号】接受到停止讯号，结束！
main结束