如何使用 .net 驱动程序 2.7.3 同时将新文档插入到 MongoDB 4 中的集合中

Question

我正在使用 MongoDB 4 和 MongoDB .net 驱动程序 2.7.3。我想同时在集合中插入一个新文档，以便集合中应该只有一个文档。文档插入集合后（即设置sequenceValue为1），我们只需要更新文档（即将sequenceValue加1），而不需要再插入新的文档。

在这个名为“countersCollection”的集合中，新文档的结构是一个 Counter 类，如下所示：

    public class Counter
    {
        [BsonId]
        public ObjectId _id { get; set; }
        public string name { get; set; }
        public long sequenceValue {get;set;}
        public DateTime date {get;set;}
    }

我的代码是这样的：

    Counter c = await this.countersCollection.AsQueryable().FirstOrDefaultAsync(x => x.name == counterName);                
    DateTime utcNow = DateTime.UtcNow;                                                                            
    if (c == null) // empty document
    {
        var options = new FindOneAndUpdateOptions<Counter, Counter>() {ReturnDocument = ReturnDocument.After, IsUpsert = true };            
        var filter = new FilterDefinitionBuilder<Counter>().Where(x => x.name == counterName);
        var update = new UpdateDefinitionBuilder<Counter>().Set(x => x.sequenceValue, 1).Set(x => x.date, utcNow);                                        
        seq = await this.countersCollection.FindOneAndUpdateAsync<Counter>(filter, update, options);                    
    }

以上代码在非并发环境下运行良好，但在并发环境下运行不佳。如果多个线程同时调用上述代码，可能会在 countersCollection 中创建多个计数器文档。

有没有办法让它同时工作。

谢谢。

Answer 1

您可以使用乐观并发方法来处理它

算法可以是：

1. 你会发现Counter 和name == counterName
2. 构建您的过滤器，例如 var filter = new FilterDefinitionBuilder<Counter>().Where(x => x.name == counterName && x.sequenceValue == c.sequenceValue );
3. 尝试查找并更新，如果结果为null重试

Answer 2

好的。让我们写一些代码。

1. 为 OptimisticConcurrency 创建单独的异常

   public class OptimisticConcurrencyException : Exception
   {
   }
   

2. 创建Counter 类

   public class Counter
   {
       [BsonId]
       public ObjectId Id { get; set; }
       public string Name { get; set; }
       public long Version { get; set; }
       public DateTime Ddate { get; set; }
   }
   

3. 创建一些简单的重试逻辑

    public class CounterRepository
    {
        private readonly IMongoCollection<Counter> _countersCollection;
   
        public CounterRepository(IMongoCollection<Counter> countersCollection)
        {
            _countersCollection = countersCollection ?? throw new ArgumentNullException(nameof(countersCollection));
        }
   
        public async Task<Counter> TryInsert(Counter counter)
        {
            var policy = Policy.Handle<OptimisticConcurrencyException>()
                .WaitAndRetryAsync(new[] {
                    TimeSpan.FromSeconds(1),
                    TimeSpan.FromSeconds(3),
                    TimeSpan.FromSeconds(7)
                });
   
            return await policy.ExecuteAsync(() => TryInsertInternal(counter));
        }
   
        private async Task<Counter> TryInsertInternal(Counter counter)
        {
            var existingCounter = await _countersCollection.Find(c => c.Id == counter.Id).FirstOrDefaultAsync();
   
            if (existingCounter == null)
                return await InsertInternal(counter);
   
            return await IncreaseVersion(existingCounter);
        }
   
        private async Task<Counter> InsertInternal(Counter counter)
        {
            try
            {
                counter.Version = 1;
                await _countersCollection.InsertOneAsync(counter);
                return counter;
            }
            // if smbd insert value after you called Find(returns null at that moment)
            // you try to insert entity with the same id and exception will be thrown
            // you try to handle it by marking with optimistic concurrency and retry policy
            // wait some time and execute the method and Find returns not null now so that
            // you will not insert new record but just increase the version
            catch (MongoException)
            {
                throw new OptimisticConcurrencyException();
            }
        }
   
        private async Task<Counter> IncreaseVersion(Counter existing)
        {
            long versionSnapshot = existing.Version;
            long nextVersion = versionSnapshot + 1;
   
            var updatedCounter = await _countersCollection.FindOneAndUpdateAsync(
                c => c.Id == existing.Id && c.Version == versionSnapshot,
                new UpdateDefinitionBuilder<Counter>().Set(c => c.Version, nextVersion));
   
            // it can be null if smbd increased existing version that you fetched from db
            // so you data is now the newest one and you throw OptimisticConcurrencyException
            if (updatedCounter == null)
                throw new OptimisticConcurrencyException();
   
            return updatedCounter;
        }
    }