Merge branch 'dev'

Author: lixizan

README.md

@@ -83,11 +83,13 @@ go test -benchmem -run=^$ -bench . github.com/lxzan/concurrency/benchmark
 goos: darwin
 goarch: arm64
 pkg: github.com/lxzan/concurrency/benchmark
-Benchmark_Fib-8          1485490               775.0 ns/op             0 B/op          0 allocs/op
-Benchmark_StdGo-8            388           3066459 ns/op          160537 B/op      10002 allocs/op
-Benchmark_Queues-8           457           2602319 ns/op          324489 B/op      11061 allocs/op
-Benchmark_Ants-8             139           7337507 ns/op          160368 B/op      10004 allocs/op
-Benchmark_GoPool-8           264           4514672 ns/op          191897 B/op      10569 allocs/op
+Benchmark_Fib-8                  1534509               775.5 ns/op             0 B/op          0 allocs/op
+Benchmark_StdGo-8                    390           3078647 ns/op          160585 B/op      10002 allocs/op
+Benchmark_QueuesSingle-8             262           4388264 ns/op          345144 B/op      10898 allocs/op
+Benchmark_QueuesMultiple-8           470           2630718 ns/op          323923 B/op      10964 allocs/op
+Benchmark_Ants-8                     178           6708482 ns/op          160374 B/op      10004 allocs/op
+Benchmark_GoPool-8                   348           3487154 ns/op          194926 B/op      10511 allocs/op
 PASS
-ok      github.com/lxzan/concurrency/benchmark  8.500s
+ok      github.com/lxzan/concurrency/benchmark  10.107s
+
 ```

benchmark/benchmark_test.go

@@ -34,7 +34,7 @@ func Benchmark_StdGo(b *testing.B) {
 	}
 }
 
-func Benchmark_Queues(b *testing.B) {
+func Benchmark_QueuesSingle(b *testing.B) {
 	q := queues.New(queues.WithConcurrency(Concurrency))
 
 	for i := 0; i < b.N; i++ {
@@ -50,6 +50,22 @@ func Benchmark_Queues(b *testing.B) {
 	}
 }
 
+func Benchmark_QueuesMultiple(b *testing.B) {
+	q := queues.New(queues.WithConcurrency(1), queues.WithMultiple(Concurrency))
+
+	for i := 0; i < b.N; i++ {
+		wg := &sync.WaitGroup{}
+		wg.Add(M)
+		for j := 0; j < M; j++ {
+			q.Push(func() {
+				fib(N)
+				wg.Done()
+			})
+		}
+		wg.Wait()
+	}
+}
+
 func Benchmark_Ants(b *testing.B) {
 	q, _ := ants.NewPool(Concurrency)
 	defer q.Release()

groups/options.go

@@ -17,9 +17,9 @@ func WithTimeout(t time.Duration) Option {
 }
 
 // WithConcurrency 设置最大并发
-func WithConcurrency(num int64) Option {
+func WithConcurrency(n int64) Option {
 	return func(o *options) {
-		o.concurrency = num
+		o.concurrency = n
 	}
 }
 

queues/multiple_queue.go

@@ -0,0 +1,73 @@
+package queues
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+)
+
+type multipleQueue struct {
+	options *options       // 参数
+	serial  int64          // 序列号
+	size    int64          // 队列大小
+	stopped atomic.Uint32  // 是否关闭
+	qs      []*singleQueue // 子队列
+}
+
+// 创建多重队列
+func newMultipleQueue(o *options) *multipleQueue {
+	qs := make([]*singleQueue, o.size)
+	for i := int64(0); i < o.size; i++ {
+		qs[i] = newSingleQueue(o)
+	}
+	return &multipleQueue{options: o, qs: qs, size: o.size}
+}
+
+// Push 追加任务
+func (c *multipleQueue) Push(job Job) {
+	if c.stopped.Load() == 0 {
+		index := atomic.AddInt64(&c.serial, 1) & (c.size - 1)
+		c.qs[index].Push(job)
+	}
+}
+
+// Stop 停止
+// 可能需要等待一段时间, 直到所有任务执行完成或者超时
+func (c *multipleQueue) Stop() error {
+	ctx, cancel := context.WithTimeout(context.Background(), c.options.timeout)
+	ticker := time.NewTicker(50 * time.Millisecond)
+	defer func() {
+		cancel()
+		ticker.Stop()
+	}()
+
+	for {
+		select {
+		case <-ticker.C:
+			if c.doStop(false) {
+				return nil
+			}
+		case <-ctx.Done():
+			c.doStop(true)
+			return ErrStopTimeout
+		}
+	}
+}
+
+func (c *multipleQueue) doStop(force bool) bool {
+	if force {
+		c.stopped.Store(1)
+		return true
+	}
+
+	sum := 0
+	for _, item := range c.qs {
+		sum += item.len()
+	}
+	if sum == 0 {
+		c.stopped.Store(1)
+		return true
+	}
+
+	return false
+}

queues/options.go

@@ -11,9 +11,9 @@ import (
 type Option func(o *options)
 
 // WithConcurrency 设置最大并发
-func WithConcurrency(num int64) Option {
+func WithConcurrency(n int64) Option {
 	return func(o *options) {
-		o.concurrency = internal.ToBinaryNumber(num)
+		o.concurrency = n
 	}
 }
 
@@ -31,6 +31,15 @@ func WithLogger(logger logs.Logger) Option {
 	}
 }
 
+// WithMultiple 设置多重队列, 降低锁竞争开销
+// 注意: n会被转化为pow(2,x)
+func WithMultiple(n int64) Option {
+	return func(o *options) {
+		o.multiple = true
+		o.size = internal.ToBinaryNumber(n)
+	}
+}
+
 // WithRecovery 设置恢复程序
 func WithRecovery() Option {
 	return func(o *options) {

queues/queue.go

@@ -1,10 +1,8 @@
 package queues
 
 import (
-	"context"
 	"github.com/lxzan/concurrency/logs"
-	"sync"
-	"sync/atomic"
+	"github.com/pkg/errors"
 	"time"
 )
 
@@ -13,10 +11,12 @@ const (
 	defaultTimeout     = 30 * time.Second
 )
 
-var DefaultQueue = New(WithConcurrency(16), WithRecovery())
+var ErrStopTimeout = errors.New("stop timeout")
 
 type (
 	options struct {
+		multiple    bool
+		size        int64
 		concurrency int64
 		timeout     time.Duration
 		caller      Caller
@@ -25,28 +25,21 @@ type (
 
 	Caller func(logger logs.Logger, f func())
 
-	queue struct {
-		mu             *sync.Mutex // 锁
-		q              []Job       // 任务队列
-		maxConcurrency int64       // 最大并发
-		curConcurrency int64       // 当前并发
-		caller         Caller      // 异常处理
-		logger         logs.Logger // 日志
-	}
-
 	Job func()
 
-	Queue struct {
-		options *options
-		serial  int64
-		qs      []*queue
+	Queue interface {
+		// 追加任务
+		Push(job Job)
+
+		// 停止
+		// 注意: 此方法有阻塞等待任务结束逻辑; 停止后调用Push方法不会产生任何效果.
+		Stop() error
 	}
 )
 
-// New
-// 创建N条并发度为1的任务队列
-func New(opts ...Option) *Queue {
+func New(opts ...Option) Queue {
 	o := &options{
+		multiple:    false,
 		concurrency: defaultConcurrency,
 		timeout:     defaultTimeout,
 		caller:      func(logger logs.Logger, f func()) { f() },
@@ -56,93 +49,8 @@ func New(opts ...Option) *Queue {
 		f(o)
 	}
 
-	qs := make([]*queue, o.concurrency)
-	for i := int64(0); i < o.concurrency; i++ {
-		qs[i] = newQueue(o)
-	}
-	return &Queue{options: o, qs: qs}
-}
-
-// Push 追加任务
-func (c *Queue) Push(job Job) {
-	index := atomic.AddInt64(&c.serial, 1) & (c.options.concurrency - 1)
-	c.qs[index].push(job)
-}
-
-// Stop 停止
-// 可能需要等待一段时间, 直到所有任务执行完成或者超时
-func (c *Queue) Stop() {
-	ctx, cancel := context.WithTimeout(context.Background(), c.options.timeout)
-	ticker := time.NewTicker(50 * time.Millisecond)
-	defer func() {
-		cancel()
-		ticker.Stop()
-	}()
-
-	for {
-		select {
-		case <-ticker.C:
-			sum := 0
-			for _, item := range c.qs {
-				sum += item.len()
-			}
-			if sum == 0 {
-				return
-			}
-		case <-ctx.Done():
-			return
-		}
-	}
-}
-
-// newQueue 创建一个任务队列
-func newQueue(o *options) *queue {
-	return &queue{
-		mu:             &sync.Mutex{},
-		maxConcurrency: 1,
-		curConcurrency: 0,
-		caller:         o.caller,
-		logger:         o.logger,
-	}
-}
-
-func (c *queue) len() int {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	return len(c.q)
-}
-
-// 获取一个任务
-func (c *queue) getJob(delta int64) 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 *queue) do(job Job) {
-	for job != nil {
-		c.caller(c.logger, job)
-		job = c.getJob(-1)
-	}
-}
-
-// push 追加任务, 有资源空闲的话会立即执行
-func (c *queue) 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)
+	if o.multiple {
+		return newMultipleQueue(o)
 	}
+	return newSingleQueue(o)
 }