StackExchange Redis 在使用异步插入/读取数据时丢失了一些密钥

Posted 2023-02-19

【中文标题】StackExchange Redis 在使用异步插入/读取数据时丢失了一些密钥

【英文标题】：StackExchange Redis some keys have been lost while using async to insert/read data

【发布时间】：2015-11-10 15:23:15

【问题描述】：

我确定我们在这里遗漏了一些非常重要的东西，所以希望有人能指出我正确的方向。提前谢谢你:)

我们目前遇到的问题：有时异步操作（读取）不会从已由异步操作写入的 db 中返回哈希值。例如，一次操作可以返回 600 个键，下一次的键数量可以是 598，下一个：596 等等。我们也遇到了同样的短集问题（当我们在集合中最多有 10 个键并在批处理中读取 10 个哈希对象时：有时我们可以得到 8 个对象，有时可以得到 6 个对象，一旦我们只得到 2 个。 我们在大约 30-40% 的操作中遇到了异步方法的问题，迁移到同步操作解决了一些情况——因为我们已经失去了性能。

我们的创建/读取批处理操作示例

protected void CreateBatch(Func<IBatch, List<Task>> action)
    
        IBatch batch = Database.CreateBatch();

        List<Task> tasks = action(batch);

        batch.Execute();

        Task.WaitAll(tasks.ToArray());
    

    protected IEnumerable<T> GetBatch<T, TRedis>(
        IEnumerable<RedisKey> keys, 
        Func<IBatch, RedisKey, Task<TRedis>> invokeBatchOperation, 
        Func<TRedis, T> buildResultItem)
    
        IBatch batch = Database.CreateBatch();
        List<RedisKey> keyList = keys.ToList();
        List<Task> tasks = new List<Task>(keyList.Count);
        List<T> result = new List<T>(keyList.Count);

        foreach (RedisKey key in keyList)
        
            Task task = invokeBatchOperation(batch, key).ContinueWith(
                t =>
                    
                        T item = buildResultItem(t.Result);
                        result.Add(item);
                    );

            tasks.Add(task);
        

        batch.Execute();
        Task.WaitAll(tasks.ToArray());

        return result;
    

接下来我们使用写操作：

private void CreateIncrementBatch(IEnumerable<DynamicDTO> dynamicDtos)
    
        CreateBatch(
            batch =>
                
                    List<Task> tasks = new List<Task>();

                    foreach (DynamicDTO dynamicDto in dynamicDtos)
                    
                        string dynamicKey = KeysBuilders.Live.Dynamic.BuildDetailsKeyByIdAndVersion(
                            dynamicDto.Id, 
                            dynamicDto.Version);
                        HashEntry[] dynamicFields = _dtoMapper.MapDynamicToHashEntries(dynamicDto);

                        Task task = batch.HashSetAsync(dynamicKey, dynamicFields, CommandFlags.HighPriority);
                        tasks.Add(task);
                    

                    return tasks;
                );
    

我们使用下一个代码示例批量读取数据

IEnumerable<RedisKey> userKeys =
                        GetIdsByUserId(userId).Select(x => (RedisKey) KeysBuilders.Live.Dynamic.BuildDetailsKeyByUserId(x));

                    return GetBatch(userKeys, (batch, key) => batch.HashGetAllAsync(key), _dtoMapper.MapToDynamic);

我们知道 batch.Execute 不是同步/不是真正的异步操作，同时我们需要稍后检查每个操作的状态。 我们确实计划对 redis 服务器进行更多的读写操作，但是使用这个问题，我们不确定我们是否走在正确的道路上）。

非常感谢任何建议/示例和指向正确方向的方法！

一些附加信息： 我们在 asp.mvc/worker 角色（.NET 版本 4.5）中使用 StackExchange redis 客户端（最新稳定版本：1.0.481）来连接和使用 Azure redis 缓存（C1，标准）。 目前，在小型测试流程中，我们在数据库中有大约 100 000 个键（主要是散列 - 基于 redis.io 中提供的建议（每个键存储多达 10 个不同对象的字段，没有大数据或文本字段存储在散列中）和集合（主要是映射，最大的一个可以占用多达 10 000 个父键））。 我们有大约 20 个小型写入器缓存（每个写入器实例写入它自己的数据子集并且不与另一个重叠，每个操作要写入的键数量最多为 100（哈希））。 此外，我们还有一个“大人物”工作人员，他可以根据当前的 redis 状态进行一些计算并将数据存储回 redis 服务器（操作量 - 每个第一个请求最多读取/写入 1200 个键，然后使用 10 000 + 键（存储和计算）。 在大人物工作的时候：没有人读写这个确切的键空间，但是小作家继续不断地写一些键。 同时，我们有许多小型读者（最多 10 万）可以请求他们特定的数据块（基于 2 个哈希实体的映射和连接。 返回给读者的哈希实体数量约为 100-500 条记录。 由于域模型中的一些限制 - 我们尝试将键存储/读取作为批处理操作（最大（最长）批处理可以有多达 500-1000 个哈希字段读取/写入缓存。我们目前不使用事务.

【参考方案1】：

也许你可以用代替

  List<T> result = new List<T>(keyList.Count);

这样的？

 ConcurrentBag<T>result = new ConcurrentBag<T>();

ConcurrentBag 表示线程安全、无序的对象集合。