Linux内核大概只有100万次每秒的收发网络数据包的能力，如果需要突破这个限制，那么在客户端发送消息的时候，需要将消息按一定的时间进行聚合再上报，那么如何实现一个像以下需求的消息队列呢？

  • 没有消息时一直阻塞，避免CPU消耗
  • 一旦有消息的时候，只从消息队列里边取最多1秒的数据
  • 或者，一旦有足够消息数量的时候，立即返回
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

package aggregate_queue

import (
	"time"
)

// 一种按时间聚合的Queue实现方式
type Queue struct {
	queue         chan interface{}
	bulkCount     int           // 按数量聚合
	aggregateTime time.Duration // 按时间聚合
}

func Create(queueLen int64, aggregate time.Duration, bulk int) *Queue {
	return &Queue{
		queue:         make(chan interface{}, queueLen),
		bulkCount:     bulk,
		aggregateTime: aggregate,
	}
}

func (q *Queue) Length() int {
	return len(q.queue)
}

func (q *Queue) Push(item interface{}) {
	q.queue <- item
}

func (q *Queue) Pops() (items []interface{}) {
	items = append(items, <-q.queue)
	t := time.After(q.aggregateTime)
	for {
		if len(items) >= q.bulkCount {
			return
		}
		select {
		case item := <-q.queue:
			items = append(items, item)
		case <-t:
			return
		}
	}
}

这里的实现方式，主要看Pops，如果没有消息，通过chan可以达到阻塞的目的（如果一直for循环读取slice的话，会导致CPU消耗非常之高），而一旦有了消息之后，立即创建一个time.After，然后通过select来继续从chan取数据，或者直到超时时间到达。

利用这种解决方式，我们实现了单机QPS超过两百万 ;-)