IM_NEW/MessageService.WebApi/Application/SquenceService.cs

using IM.Commons;
using MessageService.Infrastructure;
using Microsoft.EntityFrameworkCore;
using RedLockNet;
using StackExchange.Redis;

namespace MessageService.WebApi.Application
{
    public class SquenceService
    {
        private readonly IDatabase _database;
        private readonly IDistributedLockFactory _lockFactory;
        private readonly MessageDbContext messageDb;
        public SquenceService(IConnectionMultiplexer multiplexer,
            IDistributedLockFactory distributedLockFactory,
            MessageDbContext messageDb
            )
        {
            _database = multiplexer.GetDatabase();
            _lockFactory = distributedLockFactory;
            this.messageDb = messageDb;
        }
        public async Task<long> GetNextSquenceIdAsync(string streamKey)
        {
            string key = RedisHelper.GetSequenceIdKey(streamKey);
            string lockKey = RedisHelper.GetSequenceIdLockKey(streamKey);
            var exists = await _database.KeyExistsAsync(key);
            if (!exists)
            {
                using (var _lock = await _lockFactory.CreateLockAsync(lockKey, TimeSpan.FromSeconds(5)))
                {
                    if (_lock.IsAcquired)
                    {
                        if (!await _database.KeyExistsAsync(key))
                        {
                            var max = await messageDb.Messages
                            .Where(x => x.StreamKey == streamKey)
                            .MaxAsync(m => (long?)m.SequenceId) ?? 0;
                            await _database.StringSetAsync(key, max, TimeSpan.FromDays(7));
                        }
                    }
                }
            }
            return await _database.StringIncrementAsync(key);
        }
    }
}