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 int
	taskQueue     chan Task
	stopSignal    chan int // 停止信号，接受到数据时时停止
	waitGroup     sync.WaitGroup

	// 保证停止后不能再提交任务
	isStop bool
	mu     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)
			return
		case 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 = true
		close(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 main

import (
	"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结束