SingalR 构建推送服务器初探-白红宇

SingalR 构建推送服务器初探

阅读量：5094 次

发布时间：2019-06-13

本文共 24817 字，大约阅读时间需要 82 分钟。

　　项目需要用到推送，于是重新研究了下推送框架，最好能够独立成一个服务，与业务无关的服务，可以给所有的项目通用。找了好久最终决定用SinglR 框架。

　　Signal 是微软支持的一个运行在 Dot NET 平台上的 html websocket 框架。它出现的主要目的是实现服务器主动推送（Push）消息到客户端页面，这样客户端就不必重新发送请求或使用轮询技术来获取消息。

　　该项目中由SingalR 管理底层链接，上次逻辑单独封装。

　　ServerHub 代码如下。用来接收客户端链接。

public class ServerHub : Hub    {        private Logger logger = LogManager.GetCurrentClassLogger();        private static ServerManager severManager = new ServerManager();        public void Hello()        {            Clients.All.hello();        }        public void Test(object o)        {            logger.Info(JsonConvert.SerializeObject(o));            Clients.Client(Context.ConnectionId).RecvMessage(Context.ConnectionId, JsonConvert.SerializeObject(o));                   }        ///         /// 供客户端调用的服务器端代码        /// 
        ///         public void SendMessage(object data)        {            string dt = JsonConvert.SerializeObject(data);            logger.Info(dt);            try            {                JObject o = JObject.Parse(dt);                string moduleId = o["ModuleId"].ToString();                string[] groupIds = o["GroupId"].ToString().Split(',');                string connectionName = o["RecvName"].ToString();                string content = o["Data"].ToString();                ServerModels send = severManager.GetModelsByConnectionId(Context.ConnectionId);                string type = o["Type"].ToString();                string projectId = o.Property("ProjectId") == null || o["ProjectId"] == null || string.IsNullOrEmpty(o["ProjectId"].ToString())?send.ProjectId: o["ProjectId"].ToString();                //判断是否需要跨模块进行                foreach (string group in groupIds)                {                    List
     
       list = severManager.GetRecvConnections(projectId,moduleId, group, connectionName);                    foreach (var item in list)                    {                        try                        {                            //封装消息                            JObject obj = new JObject();                            //发送者的信息                            obj["ProjectId"] = projectId;                            obj["ModuleId"] = moduleId;                            obj["GroupId"] = group;                            obj["Type"] = type;                            obj["Content"] = content;                            obj["ServerId"] = item.ServerId;                            obj["SendName"] = send.ConnectionName;                            obj["SendType"] = send.ConnectionType;                            obj["SendOther"] = JObject.FromObject(send.Other);                            obj["RecvName"] = item.ConnectionName;                            obj["RecvType"] = item.ConnectionType;                            obj["RecvOther"] =  JObject.FromObject(item.Other);                            obj["Time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");                            if (item.IsCache == "1")                            {                                //添加消息                                severManager.AddMessageToModels(obj, item.ServerId);                                //发送消息                                Clients.Client(item.ConnectionId).RecvMessage(JsonConvert.SerializeObject(severManager.GetMessageFromModels(item.ServerId)));                            }                            else                            {                                List
      
        listObj = new List
       
        ();                                listObj.Add(obj);                                //发送消息                                Clients.Client(item.ConnectionId).RecvMessage(JsonConvert.SerializeObject(listObj));                            }                            logger.Info(JsonConvert.SerializeObject(obj)  +"connid:" + item.ConnectionId);                        }                        catch (Exception e)                        {                            //severManager.DelConnection(Context.ConnectionId, "", "");                            logger.Error(e.Message);                        }                    }                }            }            catch(Exception e)            {                logger.Error(e.Message);            }        }        public void JoinGroup(object data)        {            string dt = JsonConvert.SerializeObject(data);            logger.Info(dt);            try            {                JObject o = JObject.Parse(dt);                string projectId = o["ProjectId"].ToString();                //发送者的账号                string connectionName = o["SendName"].ToString();                string connectionType = o["SendType"].ToString();                string moduleId = o["ModuleId"].ToString();                string[] groupIds = o["GroupId"].ToString().Split(',');                string isCache = o["Config"]["IsCache"].ToString();                string content = "JoinGroup";// o["Data"].ToString();                string type = MessageType.JOINGROUP;                //是否有其他信息                object other = Utils.UtilsHelper.GetRequestData

　　ServerManager 用来管理业务逻辑。

public class ServerManager    {        private static Logger logger = LogManager.GetCurrentClassLogger();        public static List
     
       listServer = new List
      
       ();        public static List
       
         listProject = new List
        
         ();        private static bool isInited = false;        public  ServerManager()        {            //if(!isInited)            //{            //    string [] array = ConfigurationManager.AppSettings["ProjectId"].ToString().Split(',');            //    foreach (var item in array)            //    {            //        ProjectModels project = new ProjectModels();            //        project.ProjectId = item;            //        listProject.Add(project);            //        isInited = true;            //    }            //}        }        public ServerModels IsExsitServerModels(string projectId, string connectionName, string connectionType,string moduleId, string groupId)        {            return listServer.FirstOrDefault(x => x.ProjectId == projectId && x.ModuleId == moduleId && x.GroupId == groupId && x.ConnectionName == connectionName && x.ConnectionType == connectionType);        }        public bool IsExsitProjectModels(string projectId)        {            return listProject.FirstOrDefault(x => x.ProjectId == projectId) == null ? false : true;        }        public ServerModels GetModelsByConnectionId(string connectionId)        {            return listServer.FirstOrDefault(x => x.ConnectionId == connectionId);        }                /// 
                 /// 添加链接到数据库        /// 
        /// 
                 /// 
                 /// 
                 /// 
                 /// 
                 /// 
                 /// 
                 public List
         
           AddConnection(string projectId,string connectionId,string connectionName, string connectionType,string moduleId,string groupId,string isCache,object other)        {            logger.Info(string.Format("projectCode:{0},connectionId:{1},connectionName:{2},moduleId:{3},groupId:{4}",                    projectId, connectionId, connectionName, moduleId, groupId));            List
          
            list = new List
           
            (); try { //先查看是否存在这个项目 //if(!IsExsitProjectModels(projectId)) //{ // throw new Exception("没有找到这个项目"); //} ServerModels models = IsExsitServerModels(projectId, connectionName, connectionType, moduleId, groupId); ServerModels server = new ServerModels(); server.ServerId = Guid.NewGuid().ToString(); if (models != null) { server.listMessage = models.listMessage; listServer.RemoveAll(x => x.ProjectId == projectId && x.ModuleId == moduleId && x.GroupId == groupId && x.ConnectionName == connectionName && x.ConnectionType == connectionType); server.ServerId = models.ServerId; } server.ModuleId = moduleId; server.GroupId = groupId; server.ConnectionName = connectionName; server.ConnectionId = connectionId; server.ConnectionType = connectionType; server.ProjectId = projectId; server.IsCache = isCache; server.Other = other; server.CreateTime = DateTime.Now.ToString(); listServer.Add(server); list = GetRecvConnections(projectId, moduleId, groupId); } catch(Exception e) { logger.Error(e.Message); } logger.Info(list); return list; } public void AddMessageToModels(JObject content,string serverId) { ServerModels server = listServer.FirstOrDefault(x => x.ServerId == serverId); if(server != null) { listServer.FirstOrDefault(x => x.ServerId == serverId).listMessage.Add(content); } } public void ClearMessage(string serverId) { listServer.FirstOrDefault(x => x.ServerId == serverId).listMessage.Clear(); } public List
            
              GetMessageFromModels(string serverId) { return listServer.FirstOrDefault(x => x.ServerId == serverId).listMessage; } public void DelMessageFromModels(string serverId) { listServer.FirstOrDefault(x => x.ServerId == serverId).listMessage.Clear(); } /// 
              /// 发送消息 /// 
 /// 
              /// 
              /// 
              /// 
              /// 
              public List
             
               GetRecvConnections(string projectId,string moduleId,string groupId,string connectionName = "") { logger.Info(string.Format("recvConnectionIds:{0},moduleName:{1},groupName:{2}", connectionName, moduleId, groupId)); List
              
                list = new List
               
                (); try { //string projectId = listServer.FirstOrDefault(x => x.ConnectionId == connectionId).ProjectId; list = listServer.Where(x => x.ProjectId == projectId).ToList
                
                 (); if(!string.IsNullOrEmpty(moduleId)) { list = list.Where(x => x.ModuleId == moduleId).ToList
                 
                  (); } if(!string.IsNullOrEmpty(groupId)) { list = list.Where(x => x.GroupId == groupId).ToList
                  
                   (); } if(!string.IsNullOrEmpty(connectionName)) { list = list.Where(x => x.ConnectionName == connectionName).ToList
                   
                    (); } } catch (Exception e) { logger.Error(e.Message); } logger.Info(list); return list; } /// 
                     /// 清楚单个链接 /// 
 /// 
                     /// 
                     /// 
                     public List
                    
                      DelConnection(string connectionId, string moduleId, string groupId) { logger.Info(string.Format("connectionId:{0},moduleName:{1},groupName:{2}", connectionId, moduleId, groupId)); List
                     
                       list = new List
                      
                       (); try { string projectId = listServer.FirstOrDefault(x => x.ConnectionId == connectionId).ProjectId; list = GetRecvConnections(connectionId, projectId,moduleId, groupId); if(!string.IsNullOrEmpty(moduleId)) { if(!string.IsNullOrEmpty(groupId)) { listServer.RemoveAll(x => x.ConnectionId == connectionId && x.ModuleId == moduleId && x.GroupId == groupId && x.IsCache != "1"); } else { listServer.RemoveAll(x => x.ConnectionId == connectionId && x.ModuleId == moduleId && x.IsCache != "1"); } } else { listServer.RemoveAll(x => x.ConnectionId == connectionId && x.IsCache != "1"); } } catch(Exception e) { logger.Error(e.Message); } logger.Info(list); return list; } }

　　所有的分组都是由上次管理，灵活性更大一点。

　　数据结构设计如下

public class ServerModels    {        public string ServerId = string.Empty;        public string ModuleId = string.Empty;        public string GroupId = string.Empty;        public string ConnectionId = string.Empty;        public string ConnectionName = string.Empty;        public string ConnectionType = string.Empty;        public string ProjectId = string.Empty;        public string IsCache = "0";        public object Other = new object();        public string CreateTime = string.Empty;        public List
     
       listMessage = new List
      
       ();    }

　　该项目提供了这些功能：

　　1，不同项目通过ProjectId 区分，一个Project 对应多个Module .一个 Module 对应多个 Group.一个 Group 对应多个 Connection

　　2, 每个链接都能够自由选择是否缓存消息，如果缓存，那么即使离线，下次登录也会收到所有的额消息。

　　3，每个链接都可以附带本身的其他消息，给其他用户参考。

　　目前推送服务器上线给主业务提供服务，同事给两个副业务也同时提供服务。

　　其实做完了之后，发现很多地方都不是很好，稳定性是一个打的考验。目前缓存数据全部在内存，后期稳定之后会迁往redis 。发现推送相关的，还有一些优秀的框架，MQTT ，抽空也试验下。

　　这个项目是集成了android ios website server 。都使用到了pushserver。可以互相推送。

　　website 代码

define('push', ['jquery', 'common', 'signalr.core', 'signalr.hubs'], function ($, common) {    //var con = $.hubConnection('http://' + requireConfig.pageOptions["RequestAddress"] + '/pushserver/server', { useDefaultPath: false });    var con = $.hubConnection('http://192.168.2.23/pushserver/server', { useDefaultPath: false });    var chat = con.createHubProxy('ServerHub');    var push = {};    push.state = '0'    // 引用自动生成的集线器代理    var dicType = {        "0": "JoinGroup",        "1": "LevealGroup",        "2": "GetAllClients",        "11": "AddApplication",//新增申请单        "10": "BackReport",        "13": "DeleteApplication",        "14": "UpdateApplication",        "15": "ReSampled",        "16": "Sampled",        "17": "Diagnosed",//诊断完毕        "18": "UpdateApplication",//修改申请单        "19": "DeleteApplication",//删除申请单        "20": "SampleLock",//采集申请单锁定        "21": "SampleUnLock",//采集申请单解锁定        "22": "CallNumber",//叫号推送        "23": "RequireConfirm",//诊断完毕,请求审核        "24": "ReSample",//重采样推送        "25": "ReDiagnosis",//重新诊断        "26": "ReConfirm",//重新审核        "27": "RequireConsulation",//请求会诊        "28": "ConsulationComplete",//会诊完毕        "29": "RequireDiagnosis",//请求诊断        "30": "Printed",//打印完成        "31": "OverNumber",//过号        "100": "FTS6Diagnosed", //fts6医生诊断完成        "101": "FTS6AddApplication",//FTS6 增加申请单        "200": "Chat", //聊天信息    }    con.stateChanged(function (state) {        conState = state.newState;        if (state.newState != 1) {            if (!window.console) window.console = {};            if (!window.console.log) window.console.log = function () { };            //console.log("comet连接状态改变了" + state.newState);            if (conState == 1) {                window.location.reload()            }        }    });    con.error(function (error) {        console.log('SignalR error: ' + error)    });    chat.on('recvmessage', function (message) {        console.log('SignalR recved: ' + message)        // 向页面添加消息        message = JSON.parse(message)        for (var item = message.length - 1; item >= 0; item--) {            if (message[item].ServerId != "") {                chat.invoke('replymessage', message[item].ServerId)                push.Method[message[item].ModuleId](message[item].SendName, message[item].SendType, message[item].GroupId, message[item].ModuleId, dicType[message[item].Type], message[item].Content, message[item].SendOther);            }        }    });    con.start().done(function (error) {        console.log('SignalR connectioned: ' + error)        push.state = '1'    });    push.Method = {};    push.joingroup = function (projectid, moduleid, groupid, sendname, method,iscache,other) {        var data = {            ProjectId: projectid,            ModuleId: moduleid,            GroupId: groupid,            SendName: sendname,            SendType: "",            Config: {                IsCache: iscache,                Other:other            }        }        push.ProjectId = projectid;        push.ModuleId = moduleid;        push.Method[push.ModuleId] = method;        push.GroupId = groupid;        push.SendName = sendname;        push.SendType = "";        chat.invoke('joingroup', data)    }    push.sendmessage = function (moduleid, groupid, recvname, type, data, projectId) {        common.WebsiteLog("调用推送函数1");        if (!projectId) {            projectId = "";        }        var dt = {            ProjectId: projectId,            ModuleId: moduleid,            GroupId: groupid,            RecvName: recvname,            Type: type,            Data: data        }        common.WebsiteLog("调用推送函数2");        chat.invoke('sendmessage', dt)    }    push.getallclients = function (projectId,moduleId) {        var dt = {            ProjectId: projectId,            ModuleId: moduleId        }        chat.invoke('getallclients', dt)    }    push.leavelgroup = function (moduleid, groupid) {        var dt = {            ModuleId: moduleid,            GroupId: groupid,        }        chat.invoke('leavelgroup', dt)    }    console.log("complete");    return push;})

　　server 代码

public class PushClient    {        private string url = string.Empty;        private string projectCode = string.Empty;        private HubConnection _conn = null;        private IHubProxy _proxy = null;        public delegate string DelRecvMessage(string m,string content);        public DelRecvMessage recvMessage;        public bool bConn = false;        public PushClient(string url, string projectCode)        {            try            {                this.url = url;                this.projectCode = projectCode;                Init();            }            catch(Exception e)            {                LogClass.WriteLogFile("PushClient:" + e.Message);            }                    }        private void Init()        {            if (_conn != null)            {                _conn.Dispose();                _conn = null;            }            _conn = new HubConnection(url, true);            _proxy = _conn.CreateHubProxy("ServerHub");            _conn.Error += ex => _conn.Start();            _conn.Closed += () => _conn.Start();            _conn.Start();            _proxy.On
     
      ("RecvMessage", (obj) =>            {                JArray array = JArray.Parse(obj);                string serverId = array[0]["ServerId"].ToString();                _proxy.Invoke("ReplyMessage", serverId);            });            _conn.StateChanged += new Action
      
       (tgt =>            {                if (((StateChange)tgt).NewState == Microsoft.AspNet.SignalR.Client.ConnectionState.Connected)                {                    Thread.Sleep(3000);                    JoinGroup("EcgData", "", "server", "0");                    bConn = true;                }                else                {                    bConn = false;                }            });        }        public void JoinGroup(string moduleId, string groupId, string sendName, string isCache)        {            try            {                JObject o = new JObject();                o["ProjectId"] = projectCode;                o["ModuleId"] = moduleId;                o["SendType"] = "";                o["GroupId"] = groupId;                o["SendName"] = sendName;                JObject config = new JObject();                config["IsCache"] = isCache;                o["Config"] = config;                LogClass.WriteLogFile("JoinGroup:" + JsonConvert.SerializeObject(o));                _proxy.Invoke("JoinGroup", o);            }            catch(Exception e)            {                LogClass.WriteLogFile("JoinGroup:" + e.Message);            }                    }        public void LeavelGroup(string moduleId, string groupId)        {            try            {                JObject o = new JObject();                o["ModuleId"] = moduleId;                o["GroupId"] = groupId;                _proxy.Invoke("LeavelGroup", o);            }            catch(Exception e)            {                LogClass.WriteLogFile("LeavelGroup:" + e.Message);            }                   }        public void SendMessage(string moduleId, string groupId, string recvName,string type, string data)        {            try            {                JObject o = new JObject();                o["ModuleId"] = moduleId;                o["GroupId"] = groupId;                o["RecvName"] = recvName;                o["Type"] = type;                o["Data"] = data;                LogClass.WriteLogFile("SendMessage:" + o);                _proxy.Invoke("SendMessage", o);            }            catch(Exception e)            {                LogClass.WriteLogFile("SendMessage:" + e.Message);            }                    }    }

转载于:https://www.cnblogs.com/techdreaming/p/10060598.html

你可能感兴趣的文章