快捷導(dǎo)航

golang實(shí)現(xiàn)無(wú)鎖隊(duì)列的三種方式

更新時(shí)間：2026年04月14日 10:04:00 作者：游學(xué)四方

本文主要介紹了golang實(shí)現(xiàn)無(wú)鎖隊(duì)列的三種方式,包括基于CAS操作的簡(jiǎn)單有界隊(duì)列、Michael-Scott算法的無(wú)鎖鏈表隊(duì)列及Go簡(jiǎn)化指針操作方法,感興趣的可以了解一下

1. 基于 CAS 的簡(jiǎn)單有界隊(duì)列

使用固定大小的環(huán)形緩沖區(qū)，通過(guò)原子索引實(shí)現(xiàn)無(wú)鎖。

package main
import (
	"fmt"
	"sync"
	"sync/atomic"
)
// LockFreeQueue 基于 CAS 的有界無(wú)鎖隊(duì)列
type LockFreeQueue struct {
	buffer []interface{}
	head   uint64 // 讀取位置
	tail   uint64 // 寫(xiě)入位置
	cap    uint64
}
func NewLockFreeQueue(capacity int) *LockFreeQueue {
	return &LockFreeQueue{
		buffer: make([]interface{}, capacity),
		cap:    uint64(capacity),
	}
}
// Enqueue 入隊(duì)
func (q *LockFreeQueue) Enqueue(val interface{}) bool {
	for {
		tail := atomic.LoadUint64(&q.tail)
		head := atomic.LoadUint64(&q.head)
		// 隊(duì)列已滿
		if tail-head >= q.cap {
			return false
		}
		// 嘗試 CAS 更新 tail
		if atomic.CompareAndSwapUint64(&q.tail, tail, tail+1) {
			idx := tail % q.cap
			q.buffer[idx] = val
			return true
		}
		// CAS 失敗，重試
	}
}
// Dequeue 出隊(duì)
func (q *LockFreeQueue) Dequeue() (interface{}, bool) {
	for {
		head := atomic.LoadUint64(&q.head)
		tail := atomic.LoadUint64(&q.tail)
		// 隊(duì)列為空
		if head == tail {
			return nil, false
		}
		// 嘗試 CAS 更新 head
		if atomic.CompareAndSwapUint64(&q.head, head, head+1) {
			idx := head % q.cap
			val := q.buffer[idx]
			q.buffer[idx] = nil // 幫助 GC
			return val, true
		}
		// CAS 失敗，重試
	}
}
func main() {
	q := NewLockFreeQueue(100)
	var wg sync.WaitGroup
	// 生產(chǎn)者
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			for j := 0; j < 100; j++ {
				q.Enqueue(n*100 + j)
			}
		}(i)
	}
	// 消費(fèi)者
	var count int64
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				if val, ok := q.Dequeue(); ok {
					atomic.AddInt64(&count, 1)
					_ = val
				} else {
					// 空隊(duì)列時(shí)短暫休眠避免忙等
					// 實(shí)際生產(chǎn)可用 runtime.Gosched()
				}
			}
		}()
	}
	wg.Wait()
	fmt.Printf("Processed: %d\n", atomic.LoadInt64(&count))
}

2. 無(wú)界鏈表隊(duì)列（Michael-Scott 算法）

經(jīng)典的無(wú)鎖隊(duì)列算法，使用鏈表實(shí)現(xiàn)，支持動(dòng)態(tài)擴(kuò)容。

package main
import (
	"fmt"
	"sync"
	"sync/atomic"
	"unsafe"
)
// node 鏈表節(jié)點(diǎn)
type node struct {
	value interface{}
	next  unsafe.Pointer // *node
}
// LockFreeListQueue 基于 Michael-Scott 算法的無(wú)鎖隊(duì)列
type LockFreeListQueue struct {
	head unsafe.Pointer // *node
	tail unsafe.Pointer // *node
}
func NewLockFreeListQueue() *LockFreeListQueue {
	n := unsafe.Pointer(&node{})
	return &LockFreeListQueue{
		head: n,
		tail: n,
	}
}
// Enqueue 入隊(duì)
func (q *LockFreeListQueue) Enqueue(val interface{}) {
	newNode := &node{value: val}
	newNodePtr := unsafe.Pointer(newNode)
	for {
		tail := (*node)(atomic.LoadPointer(&q.tail))
		next := (*node)(atomic.LoadPointer(&tail.next))
		// 再次檢查 tail 是否變化
		if tail != (*node)(atomic.LoadPointer(&q.tail)) {
			continue
		}
		if next == nil {
			// 嘗試將新節(jié)點(diǎn)鏈接到尾部
			if atomic.CompareAndSwapPointer(&tail.next, unsafe.Pointer(nil), newNodePtr) {
				// 嘗試更新 tail 指針
				atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), newNodePtr)
				return
			}
		} else {
			// 幫助推進(jìn) tail 指針
			atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(next))
		}
	}
}
// Dequeue 出隊(duì)
func (q *LockFreeListQueue) Dequeue() (interface{}, bool) {
	for {
		head := (*node)(atomic.LoadPointer(&q.head))
		tail := (*node)(atomic.LoadPointer(&q.tail))
		next := (*node)(atomic.LoadPointer(&head.next))
		if head != (*node)(atomic.LoadPointer(&q.head)) {
			continue
		}
		if head == tail {
			if next == nil {
				return nil, false // 空隊(duì)列
			}
			// 幫助推進(jìn) tail
			atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(next))
		} else {
			val := next.value
			if atomic.CompareAndSwapPointer(&q.head, unsafe.Pointer(head), unsafe.Pointer(next)) {
				return val, true
			}
		}
	}
}
func main() {
	q := NewLockFreeListQueue()
	var wg sync.WaitGroup
	var count int64
	// 生產(chǎn)者
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			for j := 0; j < 1000; j++ {
				q.Enqueue(n*1000 + j)
			}
		}(i)
	}
	// 消費(fèi)者
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				if val, ok := q.Dequeue(); ok {
					atomic.AddInt64(&count, 1)
					_ = val
				}
			}
		}()
	}
	wg.Wait()
	fmt.Printf("Total dequeued: %d\n", atomic.LoadInt64(&count))
}

3、使用 sync/atomic 的簡(jiǎn)化版本（Go 1.19+）

Go 1.19 引入了 atomic.Pointer，可以簡(jiǎn)化指針操作：

locklessqueue.go

//locklessqueue.go
package lockless
import (
	"sync/atomic"
)
type LockFreeQueue struct {
	buf  []interface{}
	len  int32
	head int32
	tail int32
}
func NewQueue(n int32) *LockFreeQueue {
	q := &LockFreeQueue{buf: make([]interface{}, n+1, n+1), len: n + 1}
	return q
}
func (s *LockFreeQueue) PushBack(v interface{}) {
	for {
		tail := atomic.LoadInt32(&s.tail)
		n := (tail + 1) % s.len
		if atomic.CompareAndSwapInt32(&s.head, n, n) {
			continue // 隊(duì)列滿了
		}
		if !atomic.CompareAndSwapInt32(&s.tail, tail, n) {
			continue // 獲取失敗
		}
		s.buf[tail] = v
		break
	}
}
func (s *LockFreeQueue) PopFront() interface{} {
	for {
		tail := atomic.LoadInt32(&s.tail)
		head := atomic.LoadInt32(&s.head)
		if tail == head {
			continue
		}
		n := (head + 1) % s.len
		if !atomic.CompareAndSwapInt32(&s.head, head, n) {
			continue
		}
		return s.buf[head]
	}
}

測(cè)試代碼
locklessqueue_test.go

//locklessqueue_test.go
package lockless
import (
	"sync"
	"testing"
)
func TestName(t *testing.T) {
	lq := NewQueue(10)
	w := sync.WaitGroup{}
	for i := 0; i < 100; i++ {
		w.Add(1)
		go func(gi int) {
			lq.PushBack(gi)
			w.Done()
		}(i)
	}
	go func() {
		for {
			lq.PopFront()
			//	time.Sleep(1 * time.Second)
		}
	}()
	w.Wait()
}
var ch = make(chan interface{}, 50000)
func BenchmarkGo_Chan(b *testing.B) {
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		ch <- 123
		go func() {
			<-ch
		}()
	}
}
func BenchmarkGo_LockFree(b *testing.B) {
	lq := NewQueue(1000000000)
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		lq.PushBack(123)
		go func() {
			lq.PopFront()
		}()
	}
}

執(zhí)行命令

PS D:\golang\src\Test\Test> go test -v   locklessqueue_test.go locklessqueue.go     
=== RUN   TestName
--- PASS: TestName (0.00s)
PASS
ok      command-line-arguments  0.173s

性能測(cè)試

D:\golang\src\Test\Test>  go test -v -bench="." locklessqueue_test.go  locklessqueue.go
=== RUN   TestName
--- PASS: TestName (0.00s)
goos: windows
goarch: amd64
cpu: Intel(R) Core(TM) i5-9500F CPU @ 3.00GHz
BenchmarkGo_Chan
BenchmarkGo_Chan-6               2957491               414.7 ns/op
BenchmarkGo_LockFree
BenchmarkGo_LockFree-6           4356723               272.4 ns/op
PASS
ok      command-line-arguments  85.410s

到此這篇關(guān)于golang實(shí)現(xiàn)無(wú)鎖隊(duì)列的文章就介紹到這了,更多相關(guān)golang 無(wú)鎖隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: