golang 分享: 60 行代码巧妙实现一个高性能无 channel 任务队列

话不多说, 先上测试数据, 在各种负载下均有良好表现:

// small task
const (
	PoolSize   = 16
	BenchTimes = 1000
	N          = 1000
)

goos: darwin
goarch: arm64
pkg: bench
BenchmarkGwsWorkerQueue
BenchmarkGwsWorkerQueue-8   	    3302	    357841 ns/op	   55977 B/op	    2053 allocs/op
BenchmarkGopool
BenchmarkGopool-8           	    4426	    319383 ns/op	   20000 B/op	    1173 allocs/op
BenchmarkAnts
BenchmarkAnts-8             	    3026	    399899 ns/op	   16047 B/op	    1001 allocs/op
BenchmarkNbio
BenchmarkNbio-8             	    4314	    259668 ns/op	   48028 B/op	    3000 allocs/op
PASS

// medium task
const (
	PoolSize   = 16
	BenchTimes = 1000
	N          = 10000
)

goos: darwin
goarch: arm64
pkg: bench
BenchmarkGwsWorkerQueue
BenchmarkGwsWorkerQueue-8   	    1491	    808853 ns/op	   57635 B/op	    2008 allocs/op
BenchmarkGopool
BenchmarkGopool-8           	    1377	    870051 ns/op	   17266 B/op	    1029 allocs/op
BenchmarkAnts
BenchmarkAnts-8             	     886	   1324236 ns/op	   16054 B/op	    1001 allocs/op
BenchmarkNbio
BenchmarkNbio-8             	    1324	    836092 ns/op	   48000 B/op	    3000 allocs/op
PASS

// large task
const (
	PoolSize   = 16
	BenchTimes = 1000
	N          = 100000
)

goos: darwin
goarch: arm64
pkg: bench
BenchmarkGwsWorkerQueue
BenchmarkGwsWorkerQueue-8   	     193	   6026196 ns/op	   58162 B/op	    2004 allocs/op
BenchmarkGopool
BenchmarkGopool-8           	     178	   6942255 ns/op	   17108 B/op	    1019 allocs/op
BenchmarkAnts
BenchmarkAnts-8             	     174	   6300705 ns/op	   16157 B/op	    1002 allocs/op
BenchmarkNbio
BenchmarkNbio-8             	     176	   7084957 ns/op	   48071 B/op	    2995 allocs/op
PASS

测试代码 Benchmark

代码实现

package bench

import (
	"sync"
)

type (
	WorkerQueue struct {
		mu             *sync.Mutex // 锁
		q              []Job       // 任务队列
		maxConcurrency int32       // 最大并发
		curConcurrency int32       // 当前并发
	}

	Job func()
)

// NewWorkerQueue 创建一个任务队列
func NewWorkerQueue(maxConcurrency int32) *WorkerQueue {
	return &WorkerQueue{
		mu:             &sync.Mutex{},
		maxConcurrency: maxConcurrency,
		curConcurrency: 0,
	}
}

// 获取一个任务
func (c *WorkerQueue) getJob(delta int32) Job {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.curConcurrency += delta
	if c.curConcurrency >= c.maxConcurrency {
		return nil
	}
	if n := len(c.q); n == 0 {
		return nil
	}
	var result = c.q[0]
	c.q = c.q[1:]
	c.curConcurrency++
	return result
}

// 递归地执行任务
func (c *WorkerQueue) do(job Job) {
	job()
	if nextJob := c.getJob(-1); nextJob != nil {
		go c.do(nextJob)
	}
}

// Push 追加任务, 有资源空闲的话会立即执行
func (c *WorkerQueue) Push(job Job) {
	c.mu.Lock()
	c.q = append(c.q, job)
	c.mu.Unlock()
	if item := c.getJob(0); item != nil {
		go c.do(item)
	}
}

如果觉得对你有帮助, 麻烦给 gws 点个赞吧:)