使用StackExchange.Redis替换内存缓存并优化分布式锁性能

之前分布式锁方案性能差的核心原因 #

• 你之前用的RedLock.Net、RedLock-cs这类组件是为多Redis节点强一致场景设计的，每次加解锁需要多次跨节点网络通信，本身开销就大
• 你的原内存缓存方案99%的并发排队逻辑都是进程内内存级操作，没有网络开销；如果把所有排队、锁逻辑全放到Redis侧，相当于把纳秒/微秒级的内存操作换成了毫秒级的网络IO，高并发下性能必然暴跌
• 单实例部署的Azure Cache for Redis根本不需要RedLock的多节点共识逻辑，用基础的SETNX锁就足够，额外的共识逻辑完全是多余开销

优化实现（性能追平原内存缓存版本） #

核心设计思路是双层排队+双重检查：同实例内的并发请求完全复用你原来的SemaphoreSlim内存信号量排队，只有跨实例的并发请求才会走到Redis分布式锁逻辑，尽可能减少网络IO次数。
不需要引入任何第三方分布式锁组件，直接用StackExchange.Redis原生API即可，完整实现代码如下：

using System.Collections.Concurrent;
using Microsoft.Extensions.Caching.Memory;
using StackExchange.Redis;
using System.Text.Json;

public class RedisSingleFlightCache
{
    private readonly IDatabase _redisDb;
    // 短TTL本地内存缓存，拦截绝大多数热点请求
    private readonly IMemoryCache _localCache;
    // 复用原实现的本地信号量字典，同实例并发排队用
    private readonly ConcurrentDictionary<string, Lazy<SemaphoreSlim>> _localSemaphores = new();
    // 锁唯一标识，避免跨客户端误释放锁
    private readonly string _lockOwnerId = Guid.NewGuid().ToString("N");
    // Redis锁最大持有时间，根据业务接口最大耗时调整
    private const int RedisLockTimeoutMs = 10000;
    // 本地缓存默认短TTL，平衡性能与一致性
    private const int LocalCacheTtlSec = 5;

    public RedisSingleFlightCache(IConnectionMultiplexer redisConnection, IMemoryCache localCache)
    {
        _redisDb = redisConnection.GetDatabase();
        _localCache = localCache;
    }

    public async ValueTask<TValue> GetMethod<TValue>(string key, Func<Task<TValue>> dataFetcher, int redisTtlSec)
    {
        // 第一层：本地缓存检查，命中直接返回，无任何IO
        if (_localCache.TryGetValue<TValue>(key, out var localCachedValue))
            return localCachedValue;

        // 第二层：本地信号量排队，和原内存逻辑完全一致，内存级操作无开销
        var semaphore = _localSemaphores.GetOrAdd(key, _ => 
            new Lazy<SemaphoreSlim>(() => new SemaphoreSlim(1, 1))).Value;
        await semaphore.WaitAsync();
        try
        {
            // 拿到信号量后双重检查本地缓存
            if (_localCache.TryGetValue<TValue>(key, out var doubleCheckLocalVal))
                return doubleCheckLocalVal;

            // 第三层：检查Redis缓存
            var redisValue = await _redisDb.StringGetAsync(key);
            if (redisValue.HasValue)
            {
                var cachedData = JsonSerializer.Deserialize<TValue>(redisValue!);
                _localCache.Set(key, cachedData, TimeSpan.FromSeconds(LocalCacheTtlSec));
                return cachedData;
            }

            // 本地、Redis都未命中，才获取Redis分布式锁，处理跨实例并发
            var lockKey = $"lock:{key}";
            // 单实例Redis直接用原生SET NX PX实现锁，一次网络IO即可完成加锁
            bool lockAcquired = await _redisDb.LockTakeAsync(
                lockKey, 
                _lockOwnerId, 
                TimeSpan.FromMilliseconds(RedisLockTimeoutMs)
            );

            if (!lockAcquired)
            {
                // 未抢到锁则短轮询等待其他实例写入缓存值，无需重复抢锁
                await Task.Delay(20);
                var waitingValue = await _redisDb.StringGetAsync(key);
                while (!waitingValue.HasValue)
                {
                    await Task.Delay(10);
                    waitingValue = await _redisDb.StringGetAsync(key);
                }
                var waitedResult = JsonSerializer.Deserialize<TValue>(waitingValue!);
                _localCache.Set(key, waitedResult, TimeSpan.FromSeconds(LocalCacheTtlSec));
                return waitedResult;
            }

            try
            {
                // 拿到锁后最后一次双重检查Redis，避免重复执行数据获取
                var finalCheckVal = await _redisDb.StringGetAsync(key);
                if (finalCheckVal.HasValue)
                {
                    var cachedData = JsonSerializer.Deserialize<TValue>(finalCheckVal!);
                    _localCache.Set(key, cachedData, TimeSpan.FromSeconds(LocalCacheTtlSec));
                    return cachedData;
                }

                // 执行实际数据获取逻辑
                var result = await dataFetcher();
                if (result != null)
                {
                    // 写入Redis缓存
                    await _redisDb.StringSetAsync(
                        key, 
                        JsonSerializer.Serialize(result), 
                        TimeSpan.FromSeconds(redisTtlSec)
                    );
                    // 回填本地缓存
                    _localCache.Set(key, result, TimeSpan.FromSeconds(LocalCacheTtlSec));
                }
                return result;
            }
            finally
            {
                // 释放锁时校验所有者，避免误释放其他客户端的锁
                await _redisDb.LockReleaseAsync(lockKey, _lockOwnerId);
            }
        }
        finally
        {
            semaphore.Release();
            // 用完移除信号量，避免字典内存泄漏
            _localSemaphores.TryRemove(key, out _);
        }
    }
}

性能表现说明 #

同实例下1000并发同key请求的场景，性能和你原来的纯内存实现基本持平：只有第一个请求会走到Redis层，剩下999个请求全部在本地内存排队，等第一个请求完成回填本地缓存后直接返回，没有多余网络开销。跨实例并发场景下，每个实例也只会有1个请求去访问Redis抢锁，其余请求全部本地排队，网络开销比全量请求抢分布式锁的方案低2个数量级，完全满足高并发性能要求。

测试代码任务启动时机解答 #

你的测试代码里，任务在循环内调用GetMethod往valueTasks列表Add的时候就已经开始运行了，不是等Task.WaitAll才启动。
原理很简单：async方法被调用的瞬间就会开始同步执行，直到遇到第一个未完成的await操作才会挂起返回。你循环里每次调用GetMethod时，方法会立刻执行缓存检查、信号量获取等同步逻辑，直到碰到await semaphore.WaitAsync()这类等待操作时才会返回一个未完成的ValueTask，此时任务已经处于运行状态。后续的AsTask()和WaitAll只是等待所有已经启动的任务执行完成，并不是任务的启动开关。
你可以自行加日志验证：在Get()方法入口打时间戳，再在WaitAll调用前打时间戳，会发现所有Get()的调用都发生在循环执行阶段，和WaitAll没有关系。

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