Go语言高并发全局计数器实现策略与性能分析

import "sync/atomic"

// 定义一个原子计数器类型
type AtomicCounter int32

// Increment 方法原子地增加计数器的值
func (c *AtomicCounter) Increment() int32 {
    return atomic.AddInt32((*int32)(c), 1)
}

// Get 方法原子地获取计数器的当前值
func (c *AtomicCounter) Get() int32 {
    return atomic.LoadInt32((*int32)(c))
}

// 使用示例
func main() {
    var totalWorkCount AtomicCounter
    // 在多个Goroutine中调用 totalWorkCount.Increment()
    // ...
}

import "sync"

var (
    workCounters map[string]int
    mu           sync.Mutex
)

func init() {
    workCounters = make(map[string]int)
}

// IncrementTypeCounter 安全地增加指定类型的工作计数
func IncrementTypeCounter(workType string, value int) {
    mu.Lock()
    defer mu.Unlock() // 确保锁在函数返回时被释放
    workCounters[workType] += value
}

// GetTypeCounter 安全地获取指定类型的工作计数
func GetTypeCounter(workType string) int {
    mu.Lock()
    defer mu.Unlock()
    return workCounters[workType]
}

// 使用示例
func main() {
    // 在多个Goroutine中调用 IncrementTypeCounter("type1", 1)
    // ...
}

package helpers

import (
    "sync"
)

// CounterIncrementStruct 定义增量请求结构
type CounterIncrementStruct struct {
    Bucket string
    Value  int
}

// CounterQueryStruct 定义查询请求结构
type CounterQueryStruct struct {
    Bucket  string
    Channel chan int // 用于接收查询结果的Channel
}

var (
    counter              map[string]int
    counterIncrementChan chan CounterIncrementStruct
    counterQueryChan     chan CounterQueryStruct
    counterListChan      chan chan map[string]int // 用于接收所有计数列表的Channel
    once                 sync.Once // 确保初始化只执行一次
)

// CounterInitialize 初始化计数器系统
func CounterInitialize() {
    once.Do(func() {
        counter = make(map[string]int)
        // 缓冲区大小可根据实际并发情况调整
        counterIncrementChan = make(chan CounterIncrementStruct, 100) 
        counterQueryChan = make(chan CounterQueryStruct, 100)
        counterListChan = make(chan chan map[string]int, 10)
        go goCounterWriter() // 启动计数器管理Goroutine
    })
}

// goCounterWriter 是负责管理计数的Goroutine
func goCounterWriter() {
    for {
        select {
        case ci := <-counterIncrementChan:
            if len(ci.Bucket) == 0 {
                continue // 忽略空桶名
            }
            counter[ci.Bucket] += ci.Value
        case cq := <-counterQueryChan:
            val, found := counter[cq.Bucket]
            if found {
                cq.Channel <- val
            } else {
                cq.Channel <- 0 // 未找到则返回0
            }
        case cl := <-counterListChan:
            // 返回一个map的副本，防止外部直接修改内部状态
            nm := make(map[string]int)
            for k, v := range counter {
                nm[k] = v
            }
            cl <- nm
        }
    }
}

// CounterIncrement 发送一个增量请求
func CounterIncrement(bucket string, value int) {
    if len(bucket) == 0 || value == 0 {
        return
    }
    counterIncrementChan <- CounterIncrementStruct{bucket, value}
}

// CounterQuery 发送一个查询请求并等待结果
func CounterQuery(bucket string) int {
    if len(bucket) == 0 {
        return 0
    }
    reply := make(chan int) // 为每个查询创建一个临时的回复Channel
    counterQueryChan <- CounterQueryStruct{bucket, reply}
    return <-reply // 阻塞等待结果
}

// CounterList 发送一个列表请求并等待所有计数
func CounterList() map[string]int {
    reply := make(chan map[string]int)
    counterListChan <- reply
    return <-reply
}

BenchmarkChannels         100000             15560 ns/op
BenchmarkMutex   1000000              2669 ns/op