[C#]使用rx.net来帮助完成主动通知和订阅的功能

Posted 2022-09-22 petewell

[C#]使用rx.net来帮助完成主动通知和订阅的功能

前言

rxjs已经算是在前端很红的一个主题了，不过相对来说rx.net则比较少看到有人讨论，其实rx提供了很多类似functional programming的概念，甚至还提供了类似event emit的想法，也有更多处理多个异步的强大operator，而今天想要来笔记一下，如何使用rx来完成推播和订阅的需求，而这边的主题最主要会使用到的就是subject这个名词。

导览

1. 安装rx.net
2. 情境模拟
3. 完成订阅
4. 发送主动通知
5. 结论

安装rx.net

首先我们需要在visual studio安装rx.net，开起nuget并搜寻reactive的关键字，然后如下图下载

此举会同时帮你安装core和interface的部分，而我们会使用到的很多operator都是包装在Linq里面，同时也会看到很多已经熟悉的lambda语法，而如果爬文比较多看到的都是rxjs的介绍，就连官方对rx.net的说明也少得可怜，所以如果需要使用到rx.net的话，必须觉悟可能得花很多时间自行摸索，而如果对很多观念不懂的话，如果对前端熟悉的话，可以由rxjs来了解观念。

情境模拟

试想一下我们传统在做缓存数据的时候，都是用时间性来达成，但如果我们有新数据想要通知缓存更新的话，目前有另一种redis的做法，默认就有主动通知的api可以使用，但redis缓存比较适合在很多系统的共享数据情境，如果我们只是想要在各别系统缓存的话，还是放在各系统的ap上比较适合，而这时候我们其实就可以使用rx.net来达成即时通知缓存更新数据的做法。

首先我来摸拟一个会员登入数据，当有其他会员登入数据的时候，api会即时传给前端总共有哪些会员登入，先来实做一个Member.cs的dto

public class MemberModel                   
                                          
    public int Id  get; set;             
    public string Name  get; set;        
                                          

接着我们先做两笔假数据，然后使用memory cache的方式来做缓存，程序如下

    public class CacheService
    
        ObjectCache cache = MemoryCache.Default;

        public void Init()
        
            if (cache.Get("Members") == null)
            
                var Members = new List
                
                    new MemberModel Id=1,Name="kin" ,
                    new MemberModel Id=2,Name="anson" ,
                ;
                var policy = new CacheItemPolicy  Priority = CacheItemPriority.NotRemovable ;
                cache.Add("Members", Members, policy);
            
        

        public List GetMembers()
        
            return cache.Get("Members") as List;
        
    

因为我打算在系统一开始的时候，就先把缓存写进去了，所以我选择在Global.asax里面去Init缓存数据

    public class WebApiApplication : System.Web.HttpApplication
    
        protected void Application_Start()
        
            AreaRegistration.RegisterAllAreas();
            GlobalConfiguration.Configure(WebApiConfig.Register);
            FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
            RouteConfig.RegisterRoutes(RouteTable.Routes);
            CacheService service = new CacheService();
            service.Init();
            BundleConfig.RegisterBundles(BundleTable.Bundles);

        
    

接着我是使用swagger直接测试，以确保数据是有写进memory里面

    public class ValuesController : ApiController
    
        CacheService _service = new CacheService();
        // GET: api/Values
        public IHttpActionResult Get()
        
            return Ok(_service.GetMembers());
        
    

结果如下图

完成订阅

接着我们需要把写缓存的时机点，改成使用订阅的方式来实现，也就是我们去订阅Subject，而何时要去缓存的机制则改成用rx来实现，首先来看一下我们如何实做这个部分，新增一支EventSubjects.cs

    public static class EventSubjects
    
        public static Subject SubjectMembers  get; set;  = new Subject();
    

接着我希望订阅的时候一样在系统一开始就去执行，所以一样在Global.asax动手脚

    public class WebApiApplication : System.Web.HttpApplication
    
        protected void Application_Start()
        
            AreaRegistration.RegisterAllAreas();
            GlobalConfiguration.Configure(WebApiConfig.Register);
            FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
            RouteConfig.RegisterRoutes(RouteTable.Routes);
            CacheService service = new CacheService();
            service.Init();
            service.SubscribeSubjects(); //新增订阅subject
            BundleConfig.RegisterBundles(BundleTable.Bundles);

        
    

然后替CacheService新增订阅的方法

    public class CacheService
    
        ObjectCache cache = MemoryCache.Default;
        CacheItemPolicy policy = new CacheItemPolicy  Priority = CacheItemPriority.NotRemovable ;

        public void Init()
        
            if (cache.Get("Members") == null)
            
                var Members = new List
                
                    new MemberModel Id=1,Name="kin" ,
                    new MemberModel Id=2,Name="anson" ,
                ;
                cache.Add("Members", Members, policy);
            
        

       /// 

       /// 订阅的方法
       /// 
        public void SubscribeSubjects()
        
            EventSubjects.SubjectMembers.Subscribe(x =>
            
                var result = GetMembers();
                var member=result.FirstOrDefault(r => r.Id == x.Id);
                if (member == null) result.Add(x);
                cache.Set("Members", result, policy);
            );
        

        public List GetMembers()
        
            return cache.Get("Members") as List;
        
    

大功告成了，自此之后什么时候重写缓存就是用订阅的方式来实现了

发送主动通知

这边则是关键点了，当我们打web api的post时候，假设成功就去触发subject去做主动通知

    public class ValuesController : ApiController
    
        CacheService _service = new CacheService();
        // GET: api/Values
        public IHttpActionResult Get()
        
            return Ok(_service.GetMembers());
        

        public void Post(MemberModel model)
        
            EventSubjects.SubjectMembers.OnNext(model);
        
    

整个完成之后我们就可以测试看看，是否会如预期的当我新增数据的时候，数据是否有顺利写进缓存。

一开始的数据

接着打post

重新再Get一次去拉缓存数据

确认我们是使用主动通知的方式，去通知写进缓存数据了。

结论

研究rx.net的时候，文档确实很缺乏，这就造成要研究这个技术确实不是那么容易入门和学习，毕竟文档不全相关文章也不多，但相对性的来说rxjs则是有很多相关应用，而rxjs光是operator的相关应用就博大精深了，所以有兴趣的笔者就请至rxjs30天这系列文章去做一些深入了解(https://ithelp.ithome.com.tw/articles/10186104)，而这个连结更是用图示的方式让你理解各种operator的应用(http://rxmarbles.com/)，此篇没什么用到operator的部分，主要是使用到subject的特性，如果对此篇文章有任何疑问或意见，再请回复啰。

原文:大专栏  [C#]使用rx.net来帮助完成主动通知和订阅的功能