有没有更好的方法从 redis 中获取所有值？

Question

我知道从缓存中获取所有值并不是最好的主意，但我有一个内部工具的特殊用例。

我有以下代码：

public IEnumerable<KeyValuePair<K, V>> GetAllStoreEntries()
{
    var keys = RedisStore.Server.Keys(pattern: $"{GetCompletePrefix()}:*", pageSize: pageSize).ToList();
    foreach (var subList in Split(keys, pageSize))
    {
        var subKeys = subList.ToArray();
        var values = RedisStore.RedisCache.StringGet(subKeys);  <---------
        for (var i = 0; i < values.Length; i++)
        {
            var value = values[i];
            if (value.HasValue)
            {
                var svalue = JsonConvert.DeserializeObject<V>(value);
                yield return new KeyValuePair<K, V>(GetKKey(subKeys[i]), svalue);
            }
        }
    }
}

在标记位置我想使用StringGetAsync，但我不知道如何重构我可以使用它的代码。

如何重写 GetAllStoreEntries，使其并行获取拆分后的键列表？

Answer 1

我想知道您最好的选择是否可能是使用重叠管道填充字典，例如（不是确切的代码，我正在编造这个）：

async Task<Dictionary<K, V>> GetAllStoreEntries()
{
    var keys = RedisStore.Server.Keys(pattern: $"{GetCompletePrefix()}:*", pageSize: pageSize);
    var result = new Dictionary<K, V>();
    const int OVERLAP = 10; // the number of in-flight operations
    await keys.Pipe<RedisKey, RedisValue>(OVERLAP,
        key => redis.StringGet(key),
        (key, val) => result.Add(GetKey(key), Deserialize(val)));
    return result;
}

使用（未经测试）：

public static async Task Pipe<TSource, TResult>(this IEnumerable<TSource> source,
    int pipeLength,
    Func<TSource, Task<TResult>> map, Action<TSource, TResult> reduce)
{
    var buffer = new (TSource val, Task<TResult> task)[pipeLength];

    uint index = 0;
    foreach (var item in source)
    {
        var newTask = map(item);
        if (newTask.IsCompleted)
        {   // completed synchronously; check for error and move on
            reduce(item, newTask.Result);
        }
        else
        {
            var thisIndex = index++ % pipeLength;
            var oldPair = buffer[thisIndex];
            if (oldPair.task != null) reduce(oldPair.val, await oldPair.task);
            buffer[thisIndex] = (item, newTask);
        }
    }
    for (int i = 0; i < pipeLength; i++)
    {
        var oldPair = buffer[(i + index) % pipeLength];
        if (oldPair.task != null) reduce(oldPair.val, await oldPair.task);
    }
}

这个试图做的是：

• 懒惰地迭代键 - 没有ToList()
• 一次运行多个命令
• 高效地按顺序处理结果
• 作为填充字典的异步操作

完全未经测试，但是......理论上它应该可以工作！并且：由于重叠的流水线异步操作，它不会为每个键支付延迟。增加OVERLAP 可能会进一步减少延迟延迟，就像增加pageSize 一样。