golang 双端队列
虽然使用 golang 标准库 container/list 很容易实现双端队列，但经过 benchmark 压测发现数据差点意思. 在 github 中找到一个叫 edwingeng/deque 的项目，该队列的实现很不错，另外经过线上使用后，高频读写下 cpu 有下降，另外在几百万条数据堆积场景下， gc duration 也有下降. 下面分析下 deque 的设计实现.
代码地址:
https://github.com/edwingeng/deque
deque 的设计实现
优缺点
优点
- 双端队列设计
- 使用二维数组实现队列，比 list 更快，数据结构 gc 更友好
- sync.pool 优化 chunk 对象
- 更好的扩缩容机制
- 代码生成器，避免 interface{} 装箱拆箱操作.
- v2 版直接支持泛型
缺点
没锁，需要自己加一把锁.
主体设计
使用了二维数组实现了队列结构，第二层切片是一个 chunk，存放真实数据，而第一层切片关联索引第二层的 chunk.
type chunk struct {
s int
e int
data [chunkSize]Elem
}
type deque struct {
chunks []*chunk
chunkPitch []*chunk
sFree int
eFree int
}
func (dq *deque) PushFront(v Elem) {
var c *chunk
n := len(dq.chunks)
if n == 0 {
dq.expandStart()
c = dq.chunks[0]
} else {
c = dq.chunks[0] // 获取 chunks 头节点.
if c.s == 0 { // 如果该 chunk 满了， 则获取一个 chunk.
dq.expandStart() // 把 start, end 设置 243
c = dq.chunks[0]
}
}
c.s--
c.data[c.s] = v // 也把数据放到 chunk 的尾部.
}
func (dq *deque) PushBack(v Elem) {
var c *chunk
n := len(dq.chunks)
if n == 0 {
dq.expandEnd()
c = dq.chunks[n]
} else {
c = dq.chunks[n-1] // 获取尾部节点，如果该节点满了，则创建一个新的 chunk, start 和 end 为 0.
if c.e == chunkSize {
dq.expandEnd()
c = dq.chunks[n]
}
}
// 把数据放到 chunk 对应的 index 里.
c.data[c.e] = v
// 自增，求下次 index 的位置.
c.e++
}
func (dq *deque) PopBack() Elem {
n := len(dq.chunks)
if n == 0 {
return elemDefValue
}
// 获取尾部的 chunk
c := dq.chunks[n-1]
// 相等说明为空
if c.e == c.s {
return elemDefValue
}
// 获取 chunk 尾部的元素, 并向前滚动索引
c.e--
r := c.data[c.e]
// 置空
c.data[c.e] = elemDefValue
// 如果没有数据了， 则进行缩容
if c.e == 0 {
dq.shrinkEnd()
}
return r
}
func (dq *deque) Len() int {
n := len(dq.chunks)
switch n {
case 0:
return 0
case 1:
// 只有单 chunk
return dq.chunks[0].e - dq.chunks[0].s
default:
// 头部 chunk + 尾部 chunk + 中间 chunk
return chunkSize - dq.chunks[0].s + dq.chunks[n-1].e + (n-2)*chunkSize
}
}
chunk 对象缓存
使用 sync.pool 缓存 chunk 数据结构, 用来减轻 gc 开销.
var (
sharedChunkPool = newChunkPool(func() interface{} {
return &chunk{}
})
)
// 缩容
func (dq *deque) shrinkEnd() {
...
sharedChunkPool.Put(c)
...
}
// 扩容
func (dq *deque) expandEnd() {
...
c := sharedChunkPool.Get().(*chunk)
...
}
取巧的扩缩容设计
chunk 最大为 254 个元素，当超过阈值时会实例化一个新的 chunk. 每次新增的 chunk 也都会放到 chunks 切片里. deque 没有使用 list.List 链表来关联 chunk，则使用切片.
deque 对 chunks 切片做了一些优化，由于双端队列还有个向前插入 pushFront() 的场景，如果单纯用 slice，那么会不断实例化一个新切片，把新的chunk 放到 chunks[0]，以前的切片 append 后面.
不断的实例化新切片和搬迁旧切片，该操作还是有开销的. 所以抽象了一个比 chunks 更大的切片做引用占位符.
func (dq *deque) expandEnd() {
if f := dq.eFree; f == 0 {
dq.realloc()
}
// 从 sync.pool 获取新的 chunk 数组
c := sharedChunkPool.Get().(*chunk)
// 重置 start, end 偏移量
c.s, c.e = 0, 0
dq.eFree--
newEnd := len(dq.chunkPitch) - dq.eFree
dq.chunkPitch[newEnd-1] = c
// 把 chunkPitch 的数据引用到 chunks.
dq.chunks = dq.chunkPitch[dq.sFree:newEnd]
}
benchmark
PushBack/Deque<harden> 100000000 12.0 ns/op 8 B/op 0 allocs/op
PushBack/Deque 20000000 55.5 ns/op 24 B/op 1 allocs/op
PushBack/list.List 5000000 158.7 ns/op 56 B/op 1 allocs/op
PushFront/Deque<harden> 195840157 9.2 ns/op 8 B/op 0 allocs/op
PushFront/Deque 30000000 49.2 ns/op 24 B/op 1 allocs/op
PushFront/list.List 5000000 159.2 ns/op 56 B/op 1 allocs/op
Random/Deque<harden> 65623633 15.1 ns/op 0 B/op 0 allocs/op
Random/Deque 50000000 24.7 ns/op 4 B/op 0 allocs/op
Random/list.List 30000000 46.9 ns/op 28 B/op 1 allocs/op
