using Microsoft.Extensions.Logging; using ServiceStack.Messaging; using ServiceStack.Redis; using SlnMesnac.Config; using StackExchange.Redis; using System; using System.Linq; using System.Security.Cryptography; #region << 版 本 注 释 >> /*-------------------------------------------------------------------- * 版权所有 (c) 2024 WenJY 保留所有权利。 * CLR版本:4.0.30319.42000 * 机器名称:LAPTOP-E0N2L34V * 命名空间:SlnMesnac.Redis * 唯一标识:98350fd3-5b25-4395-a2f0-eb949cc4c6f6 * * 创建者:WenJY * 电子邮箱:wenjy@mesnac.com * 创建时间:2024-04-11 16:44:24 * 版本:V1.0.0 * 描述: * *-------------------------------------------------------------------- * 修改人: * 时间: * 修改说明: * * 版本:V1.0.0 *--------------------------------------------------------------------*/ #endregion << 版 本 注 释 >> namespace SlnMesnac.Redis { /// /// Redis /// public class RedisHandler { private ILogger _logger; private readonly AppConfig _appConfig; private readonly ISubscriber _subscriber; public readonly ConnectionMultiplexer redis; public RedisHandler(AppConfig appConfig, ILogger logger) { _appConfig = appConfig; redis = ConnectionMultiplexer.Connect(_appConfig.redisConfig); _subscriber = redis.GetSubscriber(); _logger = logger; } /// /// 推送消息 /// /// /// public void PublishMessage(string channel, string message) { // 生成唯一的消息ID // var messageId = $"{DateTime.UtcNow.Ticks}_{Guid.NewGuid().ToString("N")}"; // 存储消息内容 // var redisDb = redis.GetDatabase(); // redisDb.StringSet(messageId, message, TimeSpan.FromMinutes(10)); // 设置消息的有效期 // 发布消息ID long res = _subscriber.Publish(channel, message); _logger.LogInformation($"向主题:{channel};推送消息:{message};结果:{res}"); } /// /// 推送消息到指定工位的队列 /// /// 工位ID /// 消息内容 public void PublishMessageToWorker(int workerId, string message) { var channel = $"channel_worker_{workerId}"; // 生成唯一的消息ID var messageId = $"{DateTime.UtcNow.Ticks}_{Guid.NewGuid().ToString("N")}"; // 存储消息内容,使用列表结构 var redisDb = redis.GetDatabase(); redisDb.ListRightPush(channel, message); // 将消息推入对应工位的队列 _logger.LogInformation($"向工位:{workerId} 的队列推送消息ID:{messageId},消息内容:{message}"); } /// /// 指定工位,从队列中获取一条消息(客户端消费时使用) /// /// 工位ID /// 消息内容,如果没有消息则返回null public string ConsumeMessageFromWorker(string workId,int dataBaseId = 0) { var channel = workId; var redisDb = redis.GetDatabase(dataBaseId); // 从队列左侧弹出消息 var message = redisDb.ListLeftPop(channel); if (message.IsNull) { return null; } return (string)message; } /// /// 确认消息 /// /// public void AcknowledgeMessage(string messageId) { var redisDb = redis.GetDatabase(); redisDb.KeyDelete(messageId); // 删除确认的消息 _logger.LogInformation($"消息ID:{messageId} 已被确认处理"); } /// /// 订阅消息 /// /// /// public void SubscribeToChannel(string channel, Action onMessageReceived) { _subscriber.Subscribe(channel, (ch, message) => { onMessageReceived(ch, message); _logger.LogInformation($"订阅主题:{channel};收到主题:{ch};推送的消息:{message}"); }); } } }