Go 并发安全容器完全指南：从 sync.Map 到无锁数据结构

type Map struct {
    mu     Mutex

    read   atomic.Pointer[readOnly] // 只读缓存层

    dirty  map[any]*entry          // 脏数据层（需要加锁访问）

    misses int                     // 从 read 未命中次数
}

type readOnly struct {
    m       map[any]*entry
    amended bool // true if dirty map contains some key not in m
}

type entry struct {
    p atomic.Pointer[any] // 指向实际值的指针，有 nil、expunged 和有效值三种状态
}

package main

import (
    "sync"
    "testing"
)

// RWMutexMap 是用 RWMutex 保护的普通 map
type RWMutexMap struct {
    mu sync.RWMutex
    m  map[string]int
}

func NewRWMutexMap() *RWMutexMap {
    return &RWMutexMap{m: make(map[string]int)}
}

func (rm *RWMutexMap) Load(key string) (int, bool) {
    rm.mu.RLock()
    defer rm.mu.RUnlock()
    val, ok := rm.m[key]
    return val, ok
}

func (rm *RWMutexMap) Store(key string, value int) {
    rm.mu.Lock()
    defer rm.mu.Unlock()
    rm.m[key] = value
}

// SyncMap 包装 sync.Map 提供类型化接口
type SyncMap struct {
    m sync.Map
}

func (sm *SyncMap) Load(key string) (int, bool) {
    val, ok := sm.m.Load(key)
    if !ok {
        return 0, false
    }
    return val.(int), true
}

func (sm *SyncMap) Store(key string, value int) {
    sm.m.Store(key, value)
}

func BenchmarkRWMutexMapReadOnly(b *testing.B) {
    rm := NewRWMutexMap()
    for i := 0; i < 1000; i++ {
        rm.Store(string(rune('a'+i%26)), i)
    }
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            rm.Load("key")
        }
    })
}

func BenchmarkSyncMapReadOnly(b *testing.B) {
    sm := &SyncMap{}
    for i := 0; i < 1000; i++ {
        sm.Store(string(rune('a'+i%26)), i)
    }
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            sm.Load("key")
        }
    })
}

func BenchmarkRWMutexMapReadWrite(b *testing.B) {
    rm := NewRWMutexMap()
    for i := 0; i < 1000; i++ {
        rm.Store(string(rune('a'+i%26)), i)
    }
    b.RunParallel(func(pb *testing.PB) {
        i := 0
        for pb.Next() {
            if i%10 == 0 {
                rm.Store("key", i)
            } else {
                rm.Load("key")
            }
            i++
        }
    })
}

func BenchmarkSyncMapReadWrite(b *testing.B) {
    sm := &SyncMap{}
    for i := 0; i < 1000; i++ {
        sm.Store(string(rune('a'+i%26)), i)
    }
    b.RunParallel(func(pb *testing.PB) {
        i := 0
        for pb.Next() {
            if i%10 == 0 {
                sm.Store("key", i)
            } else {
                sm.Load("key")
            }
            i++
        }
    })
}

func BenchmarkRWMutexMapWriteHeavy(b *testing.B) {
    rm := NewRWMutexMap()
    b.RunParallel(func(pb *testing.PB) {
        i := 0
        for pb.Next() {
            rm.Store(string(rune('a'+i%26)), i)
            i++
        }
    })
}

func BenchmarkSyncMapWriteHeavy(b *testing.B) {
    sm := &SyncMap{}
    b.RunParallel(func(pb *testing.PB) {
        i := 0
        for pb.Next() {
            sm.Store(string(rune('a'+i%26)), i)
            i++
        }
    })
}

go test -bench=. -benchmem main_test.go

package main

import (
    "hash/fnv"
    "sync"
)

// ConcurrentMap 是分段锁实现的并发安全 map
type ConcurrentMap struct {
    segments    []*segment
    segmentMask uint32
    segmentShift uint32
}

type segment struct {
    mu   sync.RWMutex
    data map[string]any
}

// NewConcurrentMap 创建一个指定并发级别的分段锁 map
func NewConcurrentMap(concurrencyLevel int) *ConcurrentMap {
    // 确保 concurrencyLevel 是 2 的幂次
    if concurrencyLevel < 2 {
        concurrencyLevel = 2
    }
    // 向上取到最近的 2 的幂次
    sshift := 0
    ssize := 1
    for ssize < concurrencyLevel {
        sshift++
        ssize <<= 1
    }

    segments := make([]*segment, ssize)
    for i := range segments {
        segments[i] = &segment{data: make(map[string]any)}
    }

    return &ConcurrentMap{
        segments:     segments,
        segmentMask:  uint32(ssize - 1),
        segmentShift: uint32(32 - sshift),
    }
}

func (cm *ConcurrentMap) hash(key string) uint32 {
    h := fnv.New32a()
    h.Write([]byte(key))
    return h.Sum32()
}

func (cm *ConcurrentMap) getSegment(key string) *segment {
    hash := cm.hash(key)
    index := (hash >> cm.segmentShift) & cm.segmentMask
    return cm.segments[index]
}

func (cm *ConcurrentMap) Get(key string) (any, bool) {
    seg := cm.getSegment(key)
    seg.mu.RLock()
    defer seg.mu.RUnlock()
    val, ok := seg.data[key]
    return val, ok
}

func (cm *ConcurrentMap) Set(key string, value any) {
    seg := cm.getSegment(key)
    seg.mu.Lock()
    defer seg.mu.Unlock()
    seg.data[key] = value
}

func (cm *ConcurrentMap) Delete(key string) {
    seg := cm.getSegment(key)
    seg.mu.Lock()
    defer seg.mu.Unlock()
    delete(seg.data, key)
}

func (cm *ConcurrentMap) Size() int {
    total := 0
    for _, seg := range cm.segments {
        seg.mu.RLock()
        total += len(seg.data)
        seg.mu.RUnlock()
    }
    return total
}

package main

import (
    "hash/fnv"
    "sync"
)

type ConcurrentMap[K comparable, V any] struct {
    segments     []*segment[K, V]
    segmentMask  uint32
    segmentShift uint32
}

type segment[K comparable, V any] struct {
    mu   sync.RWMutex
    data map[K]V
}

func NewConcurrentMap[K comparable, V any](concurrencyLevel int) *ConcurrentMap[K, V] {
    if concurrencyLevel < 2 {
        concurrencyLevel = 2
    }
    sshift := 0
    ssize := 1
    for ssize < concurrencyLevel {
        sshift++
        ssize <<= 1
    }

    segments := make([]*segment[K, V], ssize)
    for i := range segments {
        segments[i] = &segment[K, V]{data: make(map[K]V)}
    }

    return &ConcurrentMap[K, V]{
        segments:     segments,
        segmentMask:  uint32(ssize - 1),
        segmentShift: uint32(32 - sshift),
    }
}

func (cm *ConcurrentMap[K, V]) hash(key K) uint32 {
    // 使用 Sprintf 做通用哈希，实际生产环境建议针对常用 key 类型优化
    h := fnv.New32a()
    // 泛型这里简化处理，实际可用 any(key).(interface{ Hash() uint32 }) 等策略
    // 或基于 reflect 取哈希
    return h.Sum32()
}

package main

import (
    "fmt"
    "sync"
    "time"
)

// ChannelQueue 使用 channel 实现的生产者消费者队列
type ChannelQueue struct {
    ch chan Task
}

type Task struct {
    ID   int
    Data string
}

func NewChannelQueue(capacity int) *ChannelQueue {
    return &ChannelQueue{ch: make(chan Task, capacity)}
}

func (cq *ChannelQueue) Produce(t Task) {
    cq.ch <- t
}

func (cq *ChannelQueue) Consume() (Task, bool) {
    t, ok := <-cq.ch
    return t, ok
}

func (cq *ChannelQueue) Close() {
    close(cq.ch)
}

func main() {
    queue := NewChannelQueue(100)
    var wg sync.WaitGroup

    // 3 个生产者
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < 10; j++ {
                queue.Produce(Task{ID: id*100 + j, Data: fmt.Sprintf("task-%d", j)})
            }
        }(i)
    }

    // 2 个消费者
    var consumerWg sync.WaitGroup
    for i := 0; i < 2; i++ {
        consumerWg.Add(1)
        go func(id int) {
            defer consumerWg.Done()
            for task := range queue.ch {
                fmt.Printf("Consumer %d processing task %d\n", id, task.ID)
                time.Sleep(10 * time.Millisecond)
            }
        }(i)
    }

    // 等生产者完成后关闭队列
    go func() {
        wg.Wait()
        queue.Close()
    }()

    consumerWg.Wait()
    fmt.Println("All tasks processed")
}

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

// RingBuffer 是一个无锁单生产者单消费者环形缓冲区
type RingBuffer struct {
    buffer []any
    size   uint64
    mask   uint64

    // 使用 padding 避免伪共享
    _pad0 [8]uint64
    head  uint64 // 消费者读取位置，仅消费者修改
    _pad1 [8]uint64
    tail  uint64 // 生产者写入位置，仅生产者修改
    _pad2 [8]uint64
}

func NewRingBuffer(capacity uint64) *RingBuffer {
    // 容量必须是 2 的幂次，方便用位与替代取模
    if capacity < 2 {
        capacity = 2
    }
    // 向上取到最近的 2 的幂次
    power := uint64(1)
    for power < capacity {
        power <<= 1
    }
    return &RingBuffer{
        buffer: make([]any, power),
        size:   power,
        mask:   power - 1,
        head:   0,
        tail:   0,
    }
}

// Enqueue 由生产者调用，向队列中添加元素
func (rb *RingBuffer) Enqueue(item any) bool {
    currentTail := atomic.LoadUint64(&rb.tail)
    nextTail := (currentTail + 1) & rb.mask

    // 如果队列已满（tail 的下一个位置等于 head），返回 false
    if nextTail == atomic.LoadUint64(&rb.head) {
        return false
    }

    rb.buffer[currentTail&rb.mask] = item
    atomic.StoreUint64(&rb.tail, nextTail)
    return true
}

// Dequeue 由消费者调用，从队列中取出元素
func (rb *RingBuffer) Dequeue() (any, bool) {
    currentHead := atomic.LoadUint64(&rb.head)

    // 如果队列为空
    if currentHead == atomic.LoadUint64(&rb.tail) {
        return nil, false
    }

    item := rb.buffer[currentHead&rb.mask]
    atomic.StoreUint64(&rb.head, (currentHead+1)&rb.mask)
    return item, true
}

func (rb *RingBuffer) IsEmpty() bool {
    return atomic.LoadUint64(&rb.head) == atomic.LoadUint64(&rb.tail)
}

func (rb *RingBuffer) IsFull() bool {
    return ((atomic.LoadUint64(&rb.tail)+1)&rb.mask) == atomic.LoadUint64(&rb.head)
}

func main() {
    rb := NewRingBuffer(16)

    var wg sync.WaitGroup
    wg.Add(2)

    // 生产者
    go func() {
        defer wg.Done()
        for i := 0; i < 100; i++ {
            for !rb.Enqueue(i) {
                // 队列满时自旋等待
            }
        }
    }()

    // 消费者
    go func() {
        defer wg.Done()
        count := 0
        for count < 100 {
            if item, ok := rb.Dequeue(); ok {
                fmt.Printf("Consumed: %v\n", item)
                count++
            }
        }
    }()

    wg.Wait()
    fmt.Println("Ring buffer demo completed")
}

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "unsafe"
)

// node 是栈的链表节点
type node struct {
    value int
    next  *node
}

// LockFreeStack 是无锁栈
type LockFreeStack struct {
    head atomic.Pointer[node]
}

func NewLockFreeStack() *LockFreeStack {
    return &LockFreeStack{}
}

// Push 原子地将元素压入栈顶
func (s *LockFreeStack) Push(value int) {
    newNode := &node{value: value}
    for {
        oldHead := s.head.Load()
        newNode.next = oldHead
        if s.head.CompareAndSwap(oldHead, newNode) {
            return
        }
        // CAS 失败说明其他 goroutine 修改了 head，重试
    }
}

// Pop 原子地从栈顶弹出元素
func (s *LockFreeStack) Pop() (int, bool) {
    for {
        oldHead := s.head.Load()
        if oldHead == nil {
            return 0, false
        }
        newHead := oldHead.next
        if s.head.CompareAndSwap(oldHead, newHead) {
            return oldHead.value, true
        }
        // CAS 失败说明其他 goroutine 修改了 head，重试
    }
}

func (s *LockFreeStack) IsEmpty() bool {
    return s.head.Load() == nil
}

func main() {
    stack := NewLockFreeStack()

    var wg sync.WaitGroup
    numGoroutines := 10
    itemsPerGoroutine := 100

    // 多个 goroutine 同时 push
    for i := 0; i < numGoroutines; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < itemsPerGoroutine; j++ {
                stack.Push(id*1000 + j)
            }
        }(i)
    }

    wg.Wait()

    // 统计栈中元素数量
    count := 0
    current := stack.head.Load()
    for current != nil {
        count++
        current = current.next
    }
    fmt.Printf("Total items pushed: %d, items in stack: %d\n",
        numGoroutines*itemsPerGoroutine, count)

    // 全部弹出
    popped := 0
    for {
        if _, ok := stack.Pop(); !ok {
            break
        }
        popped++
    }
    fmt.Printf("Items popped: %d\n", popped)
}

package main

import (
    "fmt"

    cmap "github.com/orcaman/concurrent-map/v2"
)

func main() {
    // 创建支持 string->any 的并发 map
    m := cmap.New[any]()

    // 存储键值对
    m.Set("key1", "value1")
    m.Set("key2", 42)

    // 读取值
    if val, ok := m.Get("key1"); ok {
        fmt.Printf("key1 = %v\n", val)
    }

    // 删除
    m.Remove("key2")

    // 使用 Upsert 原子地更新或插入
    m.Upsert("counter", 1, func(exist bool, valueInMap any, newValue any) any {
        if !exist {
            return newValue
        }
        return valueInMap.(int) + newValue.(int)
    })

    // 遍历
    for item := range m.Iter() {
        fmt.Printf("%s -> %v\n", item.Key, item.Val)
    }

    // 获取当前大小
    fmt.Printf("Size: %d\n", m.Count())
}

package main

import (
    "fmt"

    "github.com/workiva/go-datastructures/queue"
)

func main() {
    // 创建容量为 1024 的 Ring Buffer
    q := queue.NewRingBuffer(1024)

    // 批量添加元素
    items := []any{1, 2, 3, 4, 5}
    _, _ = q.GetMany(items) // 向队列中添加元素

    // 批量取出元素
    result, _ := q.GetMany(make([]any, 3))
    fmt.Printf("Dequeued: %v\n", result)
}

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    var m sync.Map

    // 并发写入
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            key := fmt.Sprintf("goroutine-%d", id)
            m.Store(key, id*10)
        }(i)
    }
    wg.Wait()

    // 读取
    if val, ok := m.Load("goroutine-5"); ok {
        fmt.Printf("goroutine-5 value: %v\n", val)
    }

    // LoadOrStore
    actual, loaded := m.LoadOrStore("goroutine-5", 999)
    fmt.Printf("LoadOrStore: value=%v, loaded=%v\n", actual, loaded)

    // Range（遍历全部）
    count := 0
    m.Range(func(key, value any) bool {
        count++
        return true // 返回 false 可提前终止遍历
    })
    fmt.Printf("Total entries: %d\n", count)

    // Delete
    m.Delete("goroutine-0")

    // 验证删除
    if _, ok := m.Load("goroutine-0"); !ok {
        fmt.Println("goroutine-0 has been deleted")
    }
}

package main

import (
    "fmt"
    "hash/fnv"
    "sync"
)

type ConcurrentMap struct {
    segments    []*segment
    segmentMask uint32
}

type segment struct {
    mu   sync.RWMutex
    data map[string]any
}

func NewConcurrentMap(numSegments int) *ConcurrentMap {
    if numSegments < 2 {
        numSegments = 2
    }
    // 向上取到 2 的幂次
    power := 1
    for power < numSegments {
        power <<= 1
    }
    segments := make([]*segment, power)
    for i := range segments {
        segments[i] = &segment{data: make(map[string]any)}
    }
    return &ConcurrentMap{
        segments:    segments,
        segmentMask: uint32(power - 1),
    }
}

func (cm *ConcurrentMap) hash(key string) uint32 {
    h := fnv.New32a()
    h.Write([]byte(key))
    return h.Sum32()
}

func (cm *ConcurrentMap) getSegment(key string) *segment {
    return cm.segments[cm.hash(key)&cm.segmentMask]
}

func (cm *ConcurrentMap) Get(key string) (any, bool) {
    seg := cm.getSegment(key)
    seg.mu.RLock()
    defer seg.mu.RUnlock()
    val, ok := seg.data[key]
    return val, ok
}

func (cm *ConcurrentMap) Set(key string, value any) {
    seg := cm.getSegment(key)
    seg.mu.Lock()
    defer seg.mu.Unlock()
    seg.data[key] = value
}

func main() {
    cm := NewConcurrentMap(16)

    var wg sync.WaitGroup
    // 并发写入
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            key := fmt.Sprintf("key-%d", id)
            cm.Set(key, id*id)
        }(i)
    }
    wg.Wait()

    // 读取
    if val, ok := cm.Get("key-50"); ok {
        fmt.Printf("key-50 = %v\n", val)
    }

    // 检查大小（近似值，非原子操作）
    total := 0
    for _, seg := range cm.segments {
        seg.mu.RLock()
        total += len(seg.data)
        seg.mu.RUnlock()
    }
    fmt.Printf("Total entries: %d\n", total)
}

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

type RingBuffer struct {
    buffer []int
    size   uint64
    mask   uint64
    head   uint64
    tail   uint64
}

func NewRingBuffer(capacity uint64) *RingBuffer {
    power := uint64(1)
    for power < capacity {
        power <<= 1
    }
    return &RingBuffer{
        buffer: make([]int, power),
        size:   power,
        mask:   power - 1,
    }
}

func (rb *RingBuffer) Enqueue(item int, timeout time.Duration) bool {
    deadline := time.Now().Add(timeout)
    for {
        currentTail := atomic.LoadUint64(&rb.tail)
        nextTail := (currentTail + 1) & rb.mask
        if nextTail == atomic.LoadUint64(&rb.head) {
            if time.Now().After(deadline) {
                return false
            }
            time.Sleep(time.Millisecond)
            continue
        }
        rb.buffer[currentTail&rb.mask] = item
        atomic.StoreUint64(&rb.tail, nextTail)
        return true
    }
}

func (rb *RingBuffer) Dequeue(timeout time.Duration) (int, bool) {
    deadline := time.Now().Add(timeout)
    for {
        currentHead := atomic.LoadUint64(&rb.head)
        if currentHead == atomic.LoadUint64(&rb.tail) {
            if time.Now().After(deadline) {
                return 0, false
            }
            time.Sleep(time.Millisecond)
            continue
        }
        item := rb.buffer[currentHead&rb.mask]
        atomic.StoreUint64(&rb.head, (currentHead+1)&rb.mask)
        return item, true
    }
}

func main() {
    rb := NewRingBuffer(8)

    var wg sync.WaitGroup
    wg.Add(2)

    go func() {
        defer wg.Done()
        for i := 0; i < 20; i++ {
            if ok := rb.Enqueue(i, time.Second); ok {
                fmt.Printf("Enqueued: %d\n", i)
            } else {
                fmt.Println("Enqueue timeout")
            }
            time.Sleep(10 * time.Millisecond)
        }
    }()

    go func() {
        defer wg.Done()
        for i := 0; i < 20; i++ {
            if val, ok := rb.Dequeue(time.Second); ok {
                fmt.Printf("Dequeued: %d\n", val)
            } else {
                fmt.Println("Dequeue timeout")
            }
            time.Sleep(15 * time.Millisecond)
        }
    }()

    wg.Wait()
}