添加Redis服务与缓存增强

wanshenmean

2026-03-02 bfd4fd8e4a05a681ec10a47992294cf752a764c4

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;
using WIDESEAWCS_RedisService.Connection;
using WIDESEAWCS_RedisService.Options;
 
namespace WIDESEAWCS_RedisService.Stream
{
    public class RedisStreamProcessingService : IStreamProcessingService
    {
        private readonly IRedisConnectionManager _connectionManager;
        private readonly RedisOptions _options;
        private readonly ILogger<RedisStreamProcessingService> _logger;
 
        public RedisStreamProcessingService(
            IRedisConnectionManager connectionManager,
            IOptions<RedisOptions> options,
            ILogger<RedisStreamProcessingService> logger)
        {
            _connectionManager = connectionManager;
            _options = options.Value;
            _logger = logger;
        }
 
        private string BuildKey(string key) => $"{_options.KeyPrefix}stream:{key}";
 
        public string AddMessage(string streamKey, Dictionary<string, string> fields)
        {
            var entries = fields.Select(f => new NameValueEntry(f.Key, f.Value)).ToArray();
            var id = _connectionManager.GetDatabase().StreamAdd(BuildKey(streamKey), entries);
            return id.ToString();
        }
 
        public bool CreateConsumerGroup(string streamKey, string groupName, string startId = "$")
        {
            try
            {
                return _connectionManager.GetDatabase()
                    .StreamCreateConsumerGroup(BuildKey(streamKey), groupName, startId, true);
            }
            catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP"))
            {
                _logger.LogWarning("消费者组 {Group} 已存在", groupName);
                return true;
            }
        }
 
        public List<StreamEntry> ReadGroup(string streamKey, string groupName, string consumerName, int count = 10)
        {
            var entries = _connectionManager.GetDatabase()
                .StreamReadGroup(BuildKey(streamKey), groupName, consumerName, ">", count);
 
            return entries?.Select(e => new StreamEntry
            {
                Id = e.Id.ToString(),
                Fields = e.Values.ToDictionary(v => v.Name.ToString(), v => v.Value.ToString())
            }).ToList() ?? new List<StreamEntry>();
        }
 
        public long Acknowledge(string streamKey, string groupName, params string[] messageIds)
        {
            var ids = messageIds.Select(id => new RedisValue(id)).ToArray();
            return _connectionManager.GetDatabase().StreamAcknowledge(BuildKey(streamKey), groupName, ids);
        }
 
        public long GetStreamLength(string streamKey)
        {
            return _connectionManager.GetDatabase().StreamLength(BuildKey(streamKey));
        }
 
        public long TrimStream(string streamKey, int maxLength)
        {
            return _connectionManager.GetDatabase().StreamTrim(BuildKey(streamKey), maxLength);
        }
    }
}