消息推送服务
服务器推送目前流行就是私信、发布/订阅等模式,基本上都是基于会话映射,消息对列等技术实现的;高性能、分布式可以如下解决:会话映射可采用redis cluster等技术实现,消息对列可使用kafka等分布式消息队列方案实现。
APM.Server基于简单
static ConcurrentDictionary<string, Session> _sessionDic = new ConcurrentDictionary<string, Session>();
和
private static ConcurrentQueue<Message> _messageQueue = new ConcurrentQueue<Message>();
实现。
部分代码如下:
/// <summary>
/// 消息转发
/// </summary>
private void ForwardMsg()
{
try
{
var msg = MessageQueue.Dequeue();
if (msg != null)
{
switch (msg.Type)
{
case (byte)MessageType.Sub:
if (!msg.IsMuti)
{
if (!SessionDic.Exists(msg.SessionID, msg.SessionID))
SessionDic.Set(this._server, msg.SessionID, msg.SessionID);
}
if (!SessionDic.Exists(msg.SessionID, msg.Sender))
SessionDic.Set(this._server, msg.Sender, msg.SessionID);
break;
case (byte)MessageType.Unsub:
if (!msg.IsMuti)
{
if (SessionDic.Exists(msg.SessionID, msg.SessionID))
SessionDic.Del(msg.SessionID, msg.SessionID);
}
if (SessionDic.Exists(msg.SessionID, msg.Sender))
SessionDic.Del(msg.Sender, msg.SessionID);
break;
default:
var session = SessionDic.Get(msg.SessionID);
if (session != null)
{
var remotes = session.UserTokenDic.List.Where(b => b.ID != msg.Sender).ToList();
if (remotes != null && remotes.Count > )
{
Parallel.For(, remotes.Count, i =>
{
this._server.SendMsg(remotes[i], Message.Serialize(msg));
});
}
}
this.OnMessage?.Invoke(msg);
break;
} }
}
catch { }
}
异步tcp通信——APM.Core 服务端概述
异步tcp通信——APM.Core 解包
异步tcp通信——APM.Server 消息推送服务的实现
异步tcp通信——APM.ConsoleDemo
转载请标明本文来源:http://www.cnblogs.com/yswenli/
更多内容欢迎star作者的github:https://github.com/yswenli/APM
如果发现本文有什么问题和任何建议,也随时欢迎交流~