go多协程、协程组以及限流
前言
因为还没有深入研究过go的协程实现机制，所以这里只是简单表述协程的使用方法。
多协程执行
func main() {
wg := &sync.WaitGroup{}
var task = printHello
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
defer wg.Done()
task()
}()
}
wg.Wait()
fmt.Println("hhh")
}
func printHello() {
now := time.Now()
fmt.Printf("Hello,now is: %d-%d-%d
", now.Year(), now.Month(), now.Day())
}
上面是使用go的协程的常用例子，使用go关键字创建新的协程执行任务，使用WaitGroup保证当所有协程都执行完毕的时候才会执行到fmt.Println("hhh")。这种方式有个弊端，由于没有控制协程数，有可能导致协程暴增。
协程组
协程组是使用一组协程执行任务的，可以看作是协程池。协程组规定了最多可以同时执行任务的协程数，避免了直接使用go关键字导致的协程暴增问题。这里先介绍第一种实现，这里要最需要注意的是Start方法中使用的ch变量。
package main
import (
"fmt"
"sync"
"time"
)
type Task func()
type Pool struct {
ConNum int // 并发数
Task chan Task // 任务channel
Wg *sync.WaitGroup // 同步组
}
func (p *Pool) init(conNum int, wg *sync.WaitGroup) {
p.ConNum = conNum
p.Wg = wg
p.Task = make(chan Task)
}
func (p *Pool) Start() {
ch := make(chan struct{}, p.ConNum) // 控制并发数
for task := range p.Task {
ch <- struct{}{}
p.Wg.Add(1)
go func() {
defer p.Wg.Done()
task()
<-ch
}()
}
}
// 添加执行任务
func (p *Pool) Execute(task Task) {
p.Task <- task
}
func main() {
pool := &Pool{}
pool.init(10, &sync.WaitGroup{})
go pool.Start()
start := time.Now()
for i := 0; i < 1000; i++ {
fn := func() {
time.Sleep(time.Millisecond)
}
pool.Execute(fn)
}
pool.Wg.Wait()
fmt.Println("tast cost: ", time.Since(start))
start = time.Now()
for i := 0; i < 1000; i++ {
time.Sleep(time.Millisecond)
}
fmt.Println("tast cost: ", time.Since(start))
}
输出：
tast cost: 11.880283ms tast cost: 1.261029362s
从上面的耗时可以看到，使用协程组确实缩短了耗时。这里使用了最多10个协程，可以对比发现，与不使用协程相比，约为其1/10，证明协程组确实有效。
协程组的第二种实现可以理解为常规的“线程池”，其中的协程是复用的。两种实现方式的耗时相当。
package main
import (
"fmt"
"sync"
"time"
)
type Task func()
type Pool struct {
ConNum int // 并发数
Task chan Task // 任务channel
Wg *sync.WaitGroup // 同步组
}
func (p *Pool) init(conNum int, wg *sync.WaitGroup) {
p.ConNum = conNum
p.Wg = wg
p.Task = make(chan Task)
}
func (p *Pool) Start() {
ch := make(chan struct{}, p.ConNum) // 控制并发数
for task := range p.Task {
ch <- struct{}{}
p.Wg.Add(1)
go func() {
defer p.Wg.Done()
task()
<-ch
}()
}
}
// 添加执行任务
func (p *Pool) Execute(task Task) {
p.Task <- task
}
type Pool2 struct {
ConNum int // 并发数
Task chan Task // 任务channel
Wg *sync.WaitGroup // 同步组
Mx *sync.Mutex
}
func (p *Pool2) init(conNum int, wg *sync.WaitGroup) {
p.ConNum = conNum
p.Wg = wg
p.Task = make(chan Task)
p.Mx = &sync.Mutex{}
}
func (p *Pool2) Start() {
go func() {
for i := 0; i < p.ConNum; i++ {
p.Wg.Add(1)
go func() {
defer p.Wg.Done()
for task := range p.Task {
task()
}
}()
}
}()
}
// 添加执行任务
func (p *Pool2) Execute(task Task) {
p.Mx.Lock()
defer p.Mx.Unlock()
if p.Task != nil {
p.Task <- task
}
}
func (p *Pool2) Done() {
p.Mx.Lock()
defer p.Mx.Unlock()
close(p.Task)
p.Task = nil
p.Wg.Wait()
}
func main() {
taskNum := 10000000
conNum := 1000
pool := &Pool{}
pool.init(conNum, &sync.WaitGroup{})
go pool.Start()
start := time.Now()
for i := 0; i < taskNum; i++ {
fn := func() {
time.Sleep(time.Millisecond)
}
pool.Execute(fn)
}
pool.Wg.Wait()
fmt.Println("pool,task cost: ", time.Since(start))
pool2 := &Pool2{}
pool2.init(conNum, &sync.WaitGroup{})
go pool2.Start()
start = time.Now()
for i := 0; i < taskNum; i++ {
fn := func() {
time.Sleep(time.Millisecond)
}
pool2.Execute(fn)
}
pool2.Done()
fmt.Println("pool2,task cost: ", time.Since(start))
}
耗时：
pool,task cost: 10.555079677s pool2,task cost: 10.513585021s
结合限流器的协程组
协程组的好处不言而喻，但高效并发执行是以资源占用为代价的，为了避免资源占用太多，可以通过go自带的限流器对可以对协程执行进行限流。如下面的代码，在使用限流器限流前，执行一百万个任务的耗时大约是1.1s，可以粗略认为QPS是百万/s，这是一个相当高的量了，所以打算用限流器进行限流，不让QPS那么高。通过设置限流器的limit参数为十万限制了任务Qps最高为10万/s，结果不出所料。
package main
import (
"context"
"fmt"
"golang.org/x/time/rate"
"sync"
"time"
)
type Task func()
type Pool struct {
ConNum int // 并发数
Task chan Task // 任务channel
Wg *sync.WaitGroup // 同步组
Limiter *rate.Limiter
Ctx context.Context
}
func (p *Pool) init(conNum int, wg *sync.WaitGroup) {
p.ConNum = conNum
p.Wg = wg
p.Task = make(chan Task)
p.Limiter = rate.NewLimiter(rate.Limit(100000), 10000)
p.Ctx = context.Background()
}
func (p *Pool) Start() {
ch := make(chan struct{}, p.ConNum) // 控制并发数
for task := range p.Task {
ch <- struct{}{}
p.Wg.Add(1)
go func() {
defer p.Wg.Done()
p.Limiter.Wait(p.Ctx)
task()
<-ch
}()
}
}
// 添加执行任务
func (p *Pool) Execute(task Task) {
p.Task <- task
}
func main() {
pool := &Pool{}
pool.init(100, &sync.WaitGroup{})
go pool.Start()
start := time.Now()
for i := 0; i < 1000000; i++ {
fn := func() {
time.Sleep(time.Nanosecond * 10)
}
pool.Execute(fn)
}
pool.Wg.Wait()
fmt.Println("tast cost: ", time.Since(start))
}
耗时：
tast cost: 9.89831723s