首页 > 资讯 > 后端开发 > 其他教程 >c#基于Redis实现轻量级消息组件的步骤

762

分享到

c#基于Redis实现轻量级消息组件的步骤

2024-04-02 19:04:59 762人浏览独家记忆

摘要

目录数据结构定义订阅通道声明消息的个性化策略处理消息自动订阅消息发送消息确认与存储问题：消息消费完没有确认机制问题：消息处理端挂了再次重启消息会丢失最后最近在开发一个轻量级asp.n

最近在开发一个轻量级asp.net mvc开发框架，需要加入日志记录，邮件发送，短信发送等功能，为了保持模块的独立性，所以需要通过消息通信的方式进行处理，为了保持框架在部署，使用，二次开发过程中的简易便捷性，所以没有选择传统的MQ，而是基于Redis的订阅发布实现一个系统内部消息组件，话不多说，上码！

数据结构定义

消息实体包含几个部分，订阅通道名称，信息头，信息体，信息差异化额外信息字典，信息头主要包含消息标识，消息日期，信息体包含信息内容，信息实体类型等


public class Message
 {
     public string MessageChannel { set; get; }
     public MessageHead @MessageHead { set; get; }
     public MessageBody @MessageBody { set; get; }
 
     [JSONExtensionData]
     public Dictionary<string,Object> @MessageExtra { set; get; }
 
     public Message()
     {
 
     }
 
     public void AddExtra(string Name, string Value)
     {
         if (@MessageExtra == null)
         {
             @MessageExtra = new Dictionary<string, object>();
         }
         @MessageExtra.Add(Name, Value);
     }
 
     public Object GetExtra(string Name)
     {
         return @MessageExtra[Name];
     }
 }
 
 public class MessageHead
 {
     public string MessageID { set; get; }
     public DateTime MessageDate { set; get; }
 
     public MessageHead()
     {
         MessageID = CommonUtil.CreateCommonGuid();
         MessageDate = DateTime.Now;
     }
 }
 
 public class MessageBody
 {
     public string MessagejsonContent { set; get; }
     public Type MessageMapperType { set; get; }
 }

注：因为消息订阅发布传递过程中，我是通过Json序列化传输的，使用过程中可能需要一些额外的键值对信息，这里在对象中定义的是Dictinary对象，但是Dictinary本身是不支持序列化的，所以需要加上注解JsonExtensionData

订阅通道声明

我们需要达到的效果是，在系统启动时，所有消息通道可以根据系统中的应用自动订阅，这里就需要一个注解来标识我们的订阅通道接收消息的实现类


[AttributeUsage(AttributeTargets.Class)]
    public class MessageChanelAttribute : Attribute
    {
        private string _ChannleName;
        public string ChannelName
        {
            get
            {
                return this._ChannleName;
            }
            set
            {
                this._ChannleName = value;
            }
 
        }
    }

消息的个性化策略处理

Redis的三方库我这里使用的是StackExchange.Redis.dll，在消息订阅时，需要为Channel指定接收到消息时的处理委托，我们在自动订阅的过程中肯定也要收集好各类消息处理类并与Channel一一对应，这时候我们就需要一个基类FastDefaultMessageHandler，我们的具体的消息处理类继承自FastDefaultMessageHandler，重写处理方法即可


[Component]
   [MessageChanelAttribute(ChannelName = "DefaultMessage")]
   public class FastDefaultMessageHandler : IFastMessageHandle
   {
       [AutoWired]
       public DBUtil @DBUtil;
 
       public void HandleMessage(RedisChannel ChannelName, RedisValue Message)
       {
           FastExecutor.Message.Design.Message Entity = JsonConvert.DeserializeObject<FastExecutor.Message.Design.Message>(Message);
           try
           {
               if (!CheckMessageIsConsume(Entity))
               {
                   this.CustomHandle(Entity);
               }
           }
           catch (Exception e)
           {
               StringBuilder ExceptionLog = new StringBuilder();
               ExceptionLog.AppendFORMat("异常Message所属Channel:{0}", Entity.MessageChannel + Environment.NewLine);
               ExceptionLog.AppendFormat("异常Message插入时间:{0}", Entity.MessageHead.MessageDate.ToString() + Environment.NewLine);
               ExceptionLog.AppendFormat("异常Message内容：{0}", Message + Environment.NewLine);
               ExceptionLog.AppendFormat("异常信息：{0}", e.Message + Environment.NewLine);
               LogUtil.WriteLog("Logs/MessageErrorLog", "log_", ExceptionLog.ToString() + Environment.NewLine);
               ExceptionLog.AppendFormat("========================================================================================================================================================================" + Environment.NewLine);
               MessageACK.MoveMessageToExceptionChannel(Entity.MessageChannel, Entity);
           }
           finally
           {
               MessageACK.ConfirmMessageFinish(Entity.MessageChannel, Entity.MessageHead.MessageID);
           }
 
       }
 
       public virtual void CustomHandle(FastExecutor.Message.Design.Message @Message)
       {
 
       }
 
       public virtual bool CheckMessageIsConsume(FastExecutor.Message.Design.Message @Message)
       {
           return false;
       }
   }

其中的HandleMessage方法就是我们在订阅Channel时对应的委托，会调用类中的CustomHandle的虚方法，子类继承重写该方法就会基于多态进行策略调用，CheckMessageIsConsume方法是用于确认消息是否重复消费的，也可以被重写，下面看一个访问日志类的实例，使用MessageChanelAttribute标注声明该实现类需要订阅发布的Channel名称为Visit，CustomHandle方法中实现了插入数据库操作，CheckMessageIsConsume方法判断该条日志数据是否已消费（已经存在于数据库）


[MessageChanelAttribute(ChannelName = "Visit")]
public class VisitLog : FastDefaultMessageHandler
{
    public override void CustomHandle(Message.Design.Message Message)
    {
        Frame_VisitLog LogEntity = JsonConvert.DeserializeObject<Frame_VisitLog>(Message.MessageBody.MessageJsonContent);
        @DBUtil.Insert(LogEntity);
        base.CustomHandle(Message);
    }
 
    public override bool CheckMessageIsConsume(Message.Design.Message Message)
    {
        Frame_VisitLog LogEntity = JsonConvert.DeserializeObject<Frame_VisitLog>(Message.MessageBody.MessageJsonContent);
        DBRow Row = new DBRow("Frame_VisitLog", "RowGuid", LogEntity.RowGuid);
        if (Row.IsExist())
        {
            return true;
        }
        else
        {
            return false;
        }
    }
}

消息自动订阅

我们希望系统在启动时就寻找出定义好Channel和实现类，自动实现订阅，这里就需要用到ioc容器，启动系统时将所有的消息处理类放入容器中，在自动订阅时全部取出来，根据消息处理类中声明的Channel名称进行自动订阅


public void Init()
      {
          List<Type> HandlerTypeList = InjectUtil.Container.GetReGIStType(typeof(IFastMessageHandle));
          foreach (Type HandlerType in HandlerTypeList)
          {
              MessageChanelAttribute Channel = Attribute.GetCustomAttribute(HandlerType, typeof(MessageChanelAttribute)) as MessageChanelAttribute;
              RedisUtil.Subscribe(Channel.ChannelName, ((FastDefaultMessageHandler)InjectUtil.Container.Resolve(HandlerType)).HandleMessage);
          }
      }

注：

1.这里的IOC容器是我自己实现的，地址：https://gitee.com/grassprogramming/FastIOC，大家可以用AutoFac代替

2.RedisUtil是对StackExchange.Redis.dll封装的处理类，地址：Https://gitee.com/grassprogramming/FastUtil

消息发送

消息只需要调用Redis的发布方法即可，将Channel名称与定义好的数据实体类传入，序列化为Json


public void SendMessage<T>(string ChannleName, T CustomMessageEntity, Dictionary<string, string> ExtraData = null)
   {
       FastExecutor.Message.Design.Message MessageEntity = new Design.Message();
       MessageEntity.MessageChannel = ChannleName;
       MessageHead Head = new MessageHead();
       MessageBody Body = new MessageBody();
       Body.MessageMapperType = typeof(T);
       Body.MessageJsonContent = JsonConvert.SerializeObject(CustomMessageEntity);
       MessageEntity.MessageHead = Head;
       MessageEntity.MessageBody = Body;
       if (ExtraData != null)
       {
           foreach (var item in ExtraData)
           {
               MessageEntity.AddExtra(item.Key, item.Value);
           }
       }
       RedisUtil.Publish(ChannleName, MessageEntity);
       MessageACK.CopyMessageToACKList(ChannleName, MessageEntity);
   }

消息确认与存储

Redis作订阅发布模式作为消息组件的问题有两方面

问题：消息消费完没有确认机制

解决方案

基于Redis的Hash存储方式建立一个消息存储字段，在发送消息时拷贝到消息Hash字典中，消费完毕后再删除，对应SendMessage中的MessageACK.CopyMessageToACKList方法和FastDefaultMessageHandler中的MessageACK.ConfirmMessageFinish方法，本质就是对Hash字典的增加与删除功能

问题：消息处理端挂了再次重启消息会丢失

解决方案

确认机制已经保证了消息即使没有被消费完但是处理端宕机消息也不会丢失，需要注意的是，消息没有丢失仅仅是Hash字典中有存储，但是消息通道中不存在了，所以我们在系统每次启动时扫描这个Hash字典，重新发布消息到Channel，这样可能导致重复消费，所以需要靠FastDefaultMessageHandler中的CheckMessageIsConsume方法判断，同时消息处理者本身处理异常我们也需要记录下来，比如发短信供应商接口有问题，消息处理异常会进入Redis的ChannelException通道，我们可以根据需求实现一个可视化界面决定是否通过手动恢复

最后

Message组件相关代码地址：https://gitee.com/grassprogramming/FastExecutor/tree/master/code/FastExecutor/FastExecutor.Message

存在不足问题：如果消息是单纯记录日志问题，没办法确认消息是否消费了

如果大家有什么好的建议，可留言一起交流学习，共同进步

以上就是C#基于Redis实现轻量级消息组件的步骤的详细内容，更多关于c#基于Redis实现消息组件的资料请关注编程网其它相关文章！

--结束END--

本文标题: c#基于Redis实现轻量级消息组件的步骤

本文链接: https://lsjlt.com/news/125257.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

回答

如何调试操作系统的错误？
操作系统

2023-11-15发布

回答

操作系统中的I/O系统是如何实现的？
操作系统

2023-11-15发布

回答

如何实现操作系统的内存管理？
操作系统

2023-11-15发布

回答

什么是虚拟内存，它对操作系统有什么影响？
操作系统

2023-11-15发布

回答

ASP中的MVC架构和WebForms架构有什么区别和使用场景？
ASP.NET

2023-11-15发布

回答

ASP中的数据验证和数据校验有什么不同？
ASP.NET

2023-11-15发布

回答

ASP中的ADO对象和DAO对象有什么区别和使用方法？
ASP.NET

2023-11-15发布

回答

Node.js中的包管理器NPM是什么？如何使用它进行依赖管理？
node.js

2023-11-15发布

回答

Vue.js中的动态组件是什么？如何使用它来动态渲染组件？
VUE

2023-11-15发布

回答

如何使用Vue.js实现懒加载和预加载？
VUE

2023-11-15发布

c#基于Redis实现轻量级消息组件的步骤

目录

数据结构定义

订阅通道声明

消息的个性化策略处理

消息自动订阅

消息发送

消息确认与存储

问题：消息消费完没有确认机制

问题：消息处理端挂了再次重启消息会丢失

最后

c#基于Redis实现轻量级消息组件的步骤

基于React封装组件的实现步骤

基于React封装组件的实现步骤是怎样的

基于WPF实现步骤控件的示例代码

ThinkPHP基于think-queue的队列插件实现消息推送

基于Canal以及消息队列实现MySQL的Binlog近实时同步

c语言怎么计算字符串长度

c++排序函数sort怎么使用

C++中beginthreadex线程启动不了怎么解决

c语言swap函数如何调用

c++中使用swap函数报错怎么解决

c++中decltype的用法是什么

c语言中sizeof语句怎么使用

C语言怎么实现对文件夹加密

c语言怎么将小数转换成整数

C++中coledatetime函数的用法是什么