Go的高并发实现：多协程和任务队列设计
前言
随着大学生活的结束，我入职了浙江的一家属于电气制造业的公司，岗位是数据分析，我算是暂时脱离了互联网这一行业，以后可能就很少接触Web开发了；把曾经对Web电子邮件系统中的性能优化的一部分抽出来记录下，也算在表达我对Web后端开发的一种敬意了
需要了解的
Golang对比其他编程语言的一大优势便是原生支持高并发，为什么说是原生支持？这就不得不提到Goroutine、Channel以及select语句
Goroutine的作用如同它自身的名称，即Go协程，它比线程更小，几十个Goroutine可能体现在底层就是五六个线程，Go语言内部帮你实现了这些Goroutine之间的内存共享，使得它比线程更易用、更高效、更轻便
Channel（一般简写为 chan）提供了一种机制，它在两个并发执行的协程之间进行同步，并通过传递与该管道元素类型相符的值来进行通信，直白的说，Channel就是用来在不同的Goroutine中交换数据的，当然，你可以将其理解为一个管道或者先进先出的队列
select语句是Go中的一个控制结构语句，类似于switch，但它只能用于通道操作，且每个case必须是一个通道操作，要么是发送要么是接收，语句会监听所有指定的通道上的操作，一旦其中一个通道准备好就会执行相应的代码块
实现思路
对于普通的需要协程辅助的操作：
1.创建任务模型Task，该模型包含一个闭包函数用以存储需要普通协程执行的操作
2.使用for创建多个Go协程
3.协程中使用select等待Channel传输任务模型Task
4.接收到任务后，调用任务模型中的闭包函数
对于频繁的数据库操作：
1.创建队列数组queue，用于存放即将操作的数据
2.使用for的时候创建多个用于数据库操作（如插入操作）的Go协程
3.在协程函数体中，也是使用select和Channel进行相互搭配
实现代码
好了，废话不多说，直接上示例代码：
package work
import (
"context"
"os"
"os/signal"
"sync"
"syscall"
"time"
)
var (
Channel = make(chan Task, 10000) // 创建一个管道，用于普通操作
DatabaseChannel = make(chan Task, 10000) // 创建一个管道，用于数据库操作
queue []interface{}
insertMutex sync.Mutex
ctx, cancel = context.WithCancel(context.Background()) // 用于监听程序退出
)
// 任务模型
type Task struct {
Value interface{}
Handler func(task Task)
Comment string
}
// 普通协程的函数体
func worker(id int) {
for {
select {
case task := <-Channel:
task.Handler(task)
case <-ctx.Done(): // 如果程序退出，停止所有协程
return
case <-time.After(time.Second): // 添加1s的延迟
}
}
}
// 初始化协程
func Init() {
// 监听程序退出
go func() {
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
<-c
cancel()
logger.Info("work.go - Cancel:", "即将退出，清空线程池")
}()
numWorkers := 5000
for i := 1; i <= numWorkers; i++ {
go worker(i) // 创建用于普通操作的协程
go databaseWorker(i) // 创建用于数据库操作的协程
}
go insertWorker() // 单独的数据库插入操作协程
}
// 数据库操作协程的函数体
func databaseWorker(id int) {
for {
select {
case task := <-DatabaseChannel:
insertMutex.Lock()
queue = append(queue, task.Value) // 插入队列
insertMutex.Unlock()
case <-ctx.Done():
return
case <-time.After(time.Second):
}
}
}
// 数据库插入协程的函数体
func insertWorker() {
insertTicker := time.NewTicker(2 * time.Second)
defer insertTicker.Stop()
for {
select {
case <-insertTicker.C: // 等待Ticker
if len(queue) > 0 {
insertMutex.Lock()
result := database.DB().CreateInBatches(queue, 100)
queue = [] // 清空队列
insertMutex.Unlock()
}
case <-ctx.Done():
return
}
}
}
// 使用
work.DatabaseChannel <- work.Task{
Value: interface{}
}
work.Channel <- work.Task{
Value: interface{},
Handler: func(task work.Task) {
// todo anything
},
}
有何提升
关于性能提升的结果，之前对电子邮件系统进行过负载测试，使用的是Vegeta测试工具批量请求系统的邮件发送接口，发送大小约为85KB的邮件到系统自身进行测试，以模拟瞬间处理大量邮件接收和发送的情况
未使用协程和任务队列时，1wQPS有>95%的错误率，使用协程和任务队列设计后，可以达到0%的错误率，TP90为6.8秒，i9-10885HK的CPU使用率为70%左右浮动，内存占用大概6.5GB，性能提升还是很明显的
最后
文章到这就结束了，也代表我的Web开发告一段落了，本项目的完整代码已在GitHub中共享：仓库地址，有需要的可以去看看😀
