如何实现支持锁获取超时的细粒度动态key级高并发锁方案

实现方案 #

核心思路 #

采用key级独立锁+自动淘汰无引用锁对象的架构，既保证完全的细粒度锁控制（只有相同key的请求才会产生锁冲突），又不会因为1000万的key空间导致内存泄漏，同时原生支持超时获取能力。

具体实现步骤 #

1. 定义锁对象和分片存储结构

首先用带缓冲channel实现支持超时获取的锁，同时给每个锁加引用计数，用于自动清理无访问的锁：

import (
    "sync"
    "sync/atomic"
    "hash/fnv"
    "time"
)

type keyLock struct {
    ch  chan struct{} // 容量为1，用于实现锁逻辑
    ref int32         // 引用计数，标记当前有多少个协程正在使用/等待这个锁
}

// 分片map，降低map操作的锁冲突，分片数设为2的幂即可，不会限制业务并发度
type lockShard struct {
    mu    sync.Mutex
    locks map[string]*keyLock
}

type KeyLocker struct {
    shards    []*lockShard
    shardMask uint32
}

2. 初始化KeyLocker

分片数选64或者128即可，仅用于降低操作锁元数据的冲突：

func NewKeyLocker(shardCount int) *KeyLocker {
    if shardCount <= 0 || (shardCount&(shardCount-1)) != 0 {
        shardCount = 64 // 默认64分片，必须是2的幂
    }
    kl := &KeyLocker{
        shards:    make([]*lockShard, shardCount),
        shardMask: uint32(shardCount - 1),
    }
    for i := range kl.shards {
        kl.shards[i] = &lockShard{
            locks: make(map[string]*keyLock),
        }
    }
    return kl
}

// 哈希函数计算key对应的分片，用fnv1a哈希性能足够
func (kl *KeyLocker) getShard(key string) *lockShard {
    h := fnv.New32a()
    h.Write([]byte(key))
    hash := h.Sum32()
    return kl.shards[hash&kl.shardMask]
}

3. 实现带超时的锁获取逻辑

func (kl *KeyLocker) Lock(key string, timeout time.Duration) bool {
    shard := kl.getShard(key)
    // 分片锁仅在操作元数据时持有，耗时极短
    shard.mu.Lock()
    l, ok := shard.locks[key]
    if !ok {
        l = &keyLock{
            ch: make(chan struct{}, 1),
        }
        l.ch <- struct{}{} // 初始化放一个元素，表示锁可用
        shard.locks[key] = l
    }
    atomic.AddInt32(&l.ref, 1) // 引用计数加1
    shard.mu.Unlock()

    // 尝试获取锁，支持超时
    select {
    case <-l.ch:
        return true
    case <-time.After(timeout):
        // 超时后回滚引用计数，检查是否需要清理锁对象
        shard.mu.Lock()
        newRef := atomic.AddInt32(&l.ref, -1)
        if newRef == 0 {
            delete(shard.locks, key)
        }
        shard.mu.Unlock()
        return false
    }
}

4. 实现锁释放逻辑

func (kl *KeyLocker) Unlock(key string) {
    shard := kl.getShard(key)
    shard.mu.Lock()
    l, ok := shard.locks[key]
    if !ok {
        shard.mu.Unlock()
        return // 异常情况，锁已经被清理
    }
    // 归还锁
    l.ch <- struct{}{}
    // 引用计数减到0则清理锁对象
    newRef := atomic.AddInt32(&l.ref, -1)
    if newRef == 0 {
        delete(shard.locks, key)
    }
    shard.mu.Unlock()
}

5. 集成到operate方法

// 全局初始化一次即可
var keyLocker = NewKeyLocker(64)

func (m objectType) operate(key string) bool {
    // 100ms超时获取锁
    if !keyLocker.Lock(key, 100*time.Millisecond) {
        return false
    }
    defer keyLocker.Unlock(key)

    // 执行业务逻辑
    return true
}

方案优势 #

• 完全满足细粒度锁要求：只有相同key的请求才会产生锁冲突，分片锁仅在操作元数据时持有，耗时在纳秒级，完全不影响业务并发
• 内存占用极低：只有当前有并发访问的key才会保留锁对象，1000万的key空间不会导致内存膨胀，常驻锁对象数量和并发请求数一致，仅几千个
• 原生支持超时获取锁，完全匹配P1需求
• 并发性能远高于分片哈希表方案：5000并发下不会有性能瓶颈，热点key也只会限制同key的并发，不会影响其他key的请求

内容的提问来源于stack exchange，提问作者DanMatlin