背景

環(huán)形緩沖器（ringr buffer）是一種用于表示一個(gè)固定尺寸、頭尾相連的緩沖區(qū)的數(shù)據(jù)結(jié)構(gòu)，適合緩存數(shù)據(jù)流。

在使用上，它就是一個(gè)固定長(zhǎng)度的FIFO隊(duì)列：

在邏輯上，我們可以把它當(dāng)成是一個(gè)環(huán)，上面有兩個(gè)指針代表當(dāng)前寫索引和讀索引：

在實(shí)現(xiàn)上，我們一般是使用一個(gè)數(shù)組去實(shí)現(xiàn)這個(gè)環(huán)，當(dāng)索引到達(dá)數(shù)組尾部的時(shí)候，則重新設(shè)置為頭部：

kfifo實(shí)現(xiàn)

kfifo是Linux內(nèi)核的隊(duì)列實(shí)現(xiàn)，它具有以下特性：

• 固定長(zhǎng)度：長(zhǎng)度是固定的，而且是向上取最小的2的平方，主要是為了實(shí)現(xiàn)快速取余。
• 無(wú)鎖：在單生產(chǎn)者和單消費(fèi)者的情況下，是不需要加鎖的。主要是因?yàn)樗饕齣n和out是不回退的，一直往前。
• 快速取余：我們都直到到達(dá)隊(duì)列末尾的時(shí)候，索引需要回退到開(kāi)頭。最簡(jiǎn)單的實(shí)現(xiàn)方式就是對(duì)索引取余，比如索引in現(xiàn)在是8，隊(duì)列長(zhǎng)度是8，in%len(q)即可回退到開(kāi)頭，但是取余操作%還是比較耗時(shí)的，因此kfifo使用in&mask實(shí)現(xiàn)快速取余，其中mask=len(q)-1。

無(wú)鎖

上面我們說(shuō)到，這個(gè)無(wú)鎖是有條件的，也就是必須在單生產(chǎn)者單消費(fèi)者情況下。這種情況下，同一時(shí)刻最多只可能會(huì)有一個(gè)寫操作和一個(gè)讀操作。但是在某一個(gè)讀操作（或?qū)懖僮鳎┑钠陂g，可能會(huì)有多個(gè)寫操作（或讀操作）發(fā)生。

因?yàn)樗饕齣n和out是不回退的，因此in一直會(huì)在out前面（或者重合）。而且in只被寫操作修改，out只被讀操作修改，因此不會(huì)沖突。

這里可能有人會(huì)擔(dān)心索引溢出的問(wèn)題，比如in到達(dá)math.MaxUint64，再+1則回到0。但是其實(shí)并不影響in和out之間的距離：

package main

import (
	"fmt"
	"math"
)

func main() {
	var in uint = math.MaxUint64
	var out uint = math.MaxUint64 - 1
	fmt.Println(in - out) // 1
	in++
	fmt.Println(in - out) // 2
	out++
	fmt.Println(in - out) // 1
}

當(dāng)然如果連續(xù)兩次溢出，就會(huì)出現(xiàn)問(wèn)題。但是由于數(shù)組長(zhǎng)度是int類型，因此也沒(méi)辦法超過(guò)math.MaxUint64，也就是in和out之間的距離最多也就是2^62，因?yàn)?code>math.MaxInt64是2^63-1，沒(méi)辦法向上取2的平方了。因此也不會(huì)出現(xiàn)溢出兩倍math.MaxUint64的情況，早在溢出之前就隊(duì)列滿了。

快速取余

前面提到取余是通過(guò)in&mask實(shí)現(xiàn)的，這有一個(gè)前提條件，也就是長(zhǎng)度必須是2的次方，因此在創(chuàng)建數(shù)組的時(shí)候，長(zhǎng)度會(huì)向上取最小的2的平方。例如一個(gè)長(zhǎng)度為8的kfifo，在二進(jìn)制表示下：

len  = 0000 1000 // 十進(jìn)制8，隊(duì)列長(zhǎng)度
mask = 0000 0111 // 十進(jìn)制7，掩碼

in   = 0000 0000 // 十進(jìn)制0，寫索引
in & mask => 0000 0000 // 十進(jìn)制0，使用 & mask
in % len  => 0000 0000 // 十進(jìn)制0，使用 % len

in         = 0000 0001 // 十進(jìn)制1，寫索引
in & mask => 0000 0001 // 十進(jìn)制1，使用 & mask
in % len  => 0000 0001 // 十進(jìn)制1，使用 % len

in         = 0000 0001 // 十進(jìn)制1，寫索引
in & mask => 0000 0001 // 十進(jìn)制1，使用 & mask
in % len  => 0000 0001 // 十進(jìn)制1，使用 % len

in         = 0000 1000 // 十進(jìn)制8，寫索引
in & mask => 0000 0000 // 十進(jìn)制0，使用 & mask
in % len  => 0000 0000 // 十進(jìn)制0，使用 % len

in         = 0001 0001 // 十進(jìn)制17，寫索引
in & mask => 0000 0001 // 十進(jìn)制1，使用 & mask
in % len  => 0000 0001 // 十進(jìn)制1，使用 % len

可以看到，使用& mask的效果是和% len一樣的。

然后我們做一個(gè)簡(jiǎn)單的性能測(cè)試：

package main

import "testing"

var (
	Len  = 8
	Mask = Len - 1
	In   = 8 - 5
)

// % len
func BenchmarkModLen(b *testing.B) {
	for i := 0; i < b.N; i++ {
		_ = In % Len
	}
}

// & Mask
func BenchmarkAndMask(b *testing.B) {
	for i := 0; i < b.N; i++ {
		_ = In & Mask
	}
}

測(cè)試結(jié)果：

BenchmarkModLen-8       1000000000               0.3434 ns/op
BenchmarkAndMask-8      1000000000               0.2520 ns/op

可以看到& mask性能確實(shí)比% len好很多，這也就是為什么要用& Mask來(lái)實(shí)現(xiàn)取余的原因了。

數(shù)據(jù)結(jié)構(gòu)

數(shù)據(jù)結(jié)構(gòu)和上面介紹的一樣，in、out標(biāo)識(shí)當(dāng)前讀寫的位置；mask是size-1，用于取索引，比%size更加高效；

type Ring[T any] struct {
	in   uint64 // 寫索引
	out  uint64 // 讀索引
	mask uint64 // 掩碼，用于取索引，代替%size
	size uint64 // 長(zhǎng)度
	data []T    // 數(shù)據(jù)
}

Push()

Push()操作很簡(jiǎn)單，首先r.in & r.mask得到寫索引，讓寫索引前進(jìn)一格，然后存入數(shù)據(jù)。

// 插入元素到隊(duì)尾
func (r *Ring[T]) Push(e T) {
	if r.Full() {
		panic("ring full")
	}
	in := r.in & r.mask
	r.in++
	r.data[in] = e
}

Pop()

Pop()操作同理，根據(jù)r.out & r.mask得到讀索引，讓讀索引前進(jìn)一格，然后讀取數(shù)據(jù)。

// 彈出隊(duì)頭元素
func (r *Ring[T]) Pop() T {
	if r.Empty() {
		panic("ring emtpy")
	}
	out := r.out & r.mask
	r.out++
	return r.data[out]
}

性能測(cè)試

Round實(shí)現(xiàn)是使用& mask，同時(shí)長(zhǎng)度會(huì)向上取2的平方；Fix實(shí)現(xiàn)是使用% size保持參數(shù)的長(zhǎng)度。

測(cè)試代碼是不斷的Push()然后Pop()：

func BenchmarkRoundPushPop(b *testing.B) {
	for i := 0; i < b.N; i++ {
		r := New[int](RoundFixSize)
		for j := 0; j < RoundFixSize; j++ {
			r.Push(j)
		}
		for j := 0; j < RoundFixSize; j++ {
			r.Pop()
		}
	}
}

測(cè)試結(jié)果：& mask的性能明顯好于% size。

BenchmarkRoundPushPop-8             2544            405621 ns/op // & mask
BenchmarkFixPushPop-8                678           1740489 ns/op // % size

無(wú)界環(huán)形緩沖器

我們可以在寫數(shù)據(jù)的時(shí)候判斷是否空間已滿，如果已滿我們可以進(jìn)行動(dòng)態(tài)擴(kuò)容，從而實(shí)現(xiàn)一個(gè)無(wú)界環(huán)形緩沖器。

Push()

在Push()時(shí)檢查到空間滿時(shí)，調(diào)用grow()擴(kuò)展空間即可：

// 插入元素到隊(duì)尾
func (r *Ring[T]) Push(e T) {
	if r.Full() {
                // 擴(kuò)展空間
		r.Grow(r.Cap() + 1)
	}
	in := r.in % r.size
	r.in++
	r.data[in] = e
}

grow()

擴(kuò)容一般是擴(kuò)展為當(dāng)前容量的兩倍，然后把原來(lái)數(shù)據(jù)copy()到新的數(shù)組，更新字段即可：

// 擴(kuò)容
func (r *Ring[T]) Grow(minSize uint64) {
	size := mmath.Max(r.size*2, minSize)
	if size > MaxSize {
		panic("size is too large")
	}
	if size < 2 {
		size = 2
	}
	// 還沒(méi)容量，直接申請(qǐng)，因?yàn)椴恍枰w移元素
	if r.size == 0 {
		r.data = make([]T, size)
		r.size = size
		return
	}
	data := make([]T, size)
	out := r.out % r.size
	len := r.Len()
	copied := copy(data[:len], r.data[out:])
	copy(data[copied:len], r.data)
	r.out = 0
	r.in = len
	r.size = size
	r.data = data
}

線程安全性

由于可能會(huì)動(dòng)態(tài)擴(kuò)容，需要修改out、in指針，因此需要加鎖保證安全。

代碼地址

https://github.com/jiaxwu/gommon/tree/main/container/ringbuffer

到此這篇關(guān)于詳解Golang如何實(shí)現(xiàn)一個(gè)環(huán)形緩沖器的文章就介紹到這了,更多相關(guān)Golang環(huán)形緩沖器內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論