Azure Redis 缓存：读取器完成后不允许读取

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：读取器完成后不允许读取。 在 System.IO.Pipelines.ThrowHelper.ThrowInvalidOperationException_NoReadingAllowed() 在 System.IO.Pipelines.Pipe.AdvanceReader（SequencePosition& 消耗，SequencePosition& 检查） 在 System.IO.Pipelines.Pipe.DefaultPipeReader.AdvanceTo（消耗的 SequencePosition，检查的 SequencePosition） 在 /_/src/StackExchange.Redis/PhysicalConnection.cs 中的 StackExchange.Redis.PhysicalConnection.ReadFromPipe() 处：第 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)

   这是后台服务的主循环。只要取消标记 (

   stoppingToken

   ) 尚未收到取消信号，它就会继续执行。取消令牌通常用于在应用程序关闭时正常停止后台操作。

2. await Task.Delay(1000, stoppingToken);

   在循环内部，这行代码在循环迭代之间引入了 1000 毫秒（1 秒）的延迟。

   await

   关键字表示此操作是异步的，并允许在发生延迟时并发执行其他任务。