c# - Azure Redis 缓存 : Reading is not allowed after reader was completed

Question

我编写了缓存更新功能。

这是我的 Redis 缓存实现:

    public class RedisCache : ICache
    {
        private readonly IDatabase _cache;

        public RedisCache(string connectionString)
        {
            var connection = ConnectionMultiplexer.Connect(connectionString);
            _cache = connection.GetDatabase();
        }

        public async Task<T> GetAsync<T>(string key)
        {
            var value = await _cache.StringGetAsync(key);

            if (value.HasValue)
            {
                return JsonConvert.DeserializeObject<T>(value);
            }

            return default(T);
        }

        public async Task SetAsync<T>(string key, T value, TimeSpan ttl)
        {
            var serializedValue = JsonConvert.SerializeObject(value);

            await _cache.StringSetAsync(key, serializedValue, ttl);
        }

        public T Get<T>(string key)
        {
            var value = _cache.StringGet(key);

            if (value.HasValue)
            {
                return JsonConvert.DeserializeObject<T>(value);
            }

            return default(T);
        }

        public void Set<T>(string key, T value, TimeSpan ttl)
        {
            var serializedValue = JsonConvert.SerializeObject(value);

            _cache.StringSet(key, serializedValue, ttl);
        }

        public async Task RemoveAsync(string key)
        {
            await _cache.KeyDeleteAsync(key);
        }
    }

以及BackgroundService中的用法:

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            using var scope = _serviceProvider.CreateScope();
            _cache = scope.ServiceProvider.GetService<ICache>();
            _queue = scope.ServiceProvider.GetService<RefresBackgroundQueue>();
            
            await foreach (string item in _queue.DequeueAllAsync(stoppingToken))
            {
                try
                {
                    // some logic
                    await _cache.GetAsync<Model>(key)
                    // some logic
                    await _cache.SetAsync(randomKey, result, new TimeSpan(100, 0, 0, 0));
                }
                catch (Exception ex)
                {
                    _logger.LogError($"Exception while updating cache: {ex.Message}, {ex.StackTrace}");
                }
            }
        }

从上面的代码中我得到了无法捕获的异常:

System.InvalidOperationException: Reading is not allowed after reader was completed. at System.IO.Pipelines.ThrowHelper.ThrowInvalidOperationException_NoReadingAllowed() at System.IO.Pipelines.Pipe.AdvanceReader(SequencePosition& consumed, SequencePosition& examined) at System.IO.Pipelines.Pipe.DefaultPipeReader.AdvanceTo(SequencePosition consumed, SequencePosition examined) at StackExchange.Redis.PhysicalConnection.ReadFromPipe() in /_/src/StackExchange.Redis/PhysicalConnection.cs:line 1729

遇到这种情况我该怎么办？

Answer 1

我在 BackgroundService.cs 中的 ExecuteAsync 中做了一些更改，
代码运行良好

ExecuteAsync 中的更改

完整代码

  using Microsoft.Extensions.DependencyInjection;
  using Microsoft.Extensions.Hosting;
  using Microsoft.Extensions.Logging;
  using System;
  using System.Diagnostics;
  using System.Threading;
  using System.Threading.Tasks;
  using static Microsoft.EntityFrameworkCore.DbLoggerCategory;
    
    public class MyBackgroundService : BackgroundService
    {
        private readonly IServiceProvider _serviceProvider;
        private readonly ILogger<MyBackgroundService> _logger;
    
        public MyBackgroundService(IServiceProvider serviceProvider, ILogger<MyBackgroundService> logger)
        {
            _serviceProvider = serviceProvider;
            _logger = logger;
        }
    
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            _logger.LogInformation("MyBackgroundService is starting.");
    
            while (!stoppingToken.IsCancellationRequested)
            {
                // Your background service logic here
    
                await Task.Delay(1000, stoppingToken); // Simulate some work
            }
    
            _logger.LogInformation("MyBackgroundService is stopping.");
    
            using var scope = _serviceProvider.CreateScope();
            var cache = scope.ServiceProvider.GetRequiredService<ICache>();
            var queue = scope.ServiceProvider.GetRequiredService<RefreshBackgroundQueue>();
    
            await foreach (string item in queue.DequeueAllAsync(stoppingToken))
            {
                try
                {
                    // Your logic here
                    var key = item; 
    
                    // Retrieve data from cache
                    var model = await cache.GetAsync<Model>(key);
    
                    // Perform some processing
                    var processedResult = ProcessModel(model);
    
                    // Store result in cache
                    var randomKey = Guid.NewGuid().ToString();
    
                    await cache.SetAsync(randomKey, processedResult, TimeSpan.FromDays(1)); // Adjust the TTL as needed
                }
                catch (Exception ex)
                {
                    _logger.LogError($"Exception while updating cache: {ex.Message}, {ex.StackTrace}");
                }
            }
        }

在代码中我使用了以下条件

1. while (!stoppingToken.IsCancellationRequested) 这是后台服务的主循环。只要取消 token (stoppingToken) 尚未收到取消信号，它就会继续执行。取消 token 通常用于在应用程序关闭时正常停止后台操作。

2. await Task.Delay(1000,stoppingToken); 在循环内部，这行代码在循环迭代之间引入了 1000 毫秒(1 秒)的延迟。 await 关键字表示此操作是异步的，并允许其他任务在发生延迟时并发执行。