C# socket 实现消息中心向消息平台 转发消息
2024-08-24 06:51:56
公司用到,直接粘代码了
using System; using System.Collections.Generic; using System.Configuration; using System.Linq; using System.Net; using System.Net.Sockets; using System.Text; using System.Threading; using System.Threading.Tasks; using Jinher.AMP.SNS.Chat.Client; using Jinher.AMP.SNS.Chat.Deploy.CustomDTO; using Jinher.AMP.SNS.Chat.Service; using Jinher.AMP.SNS.Chat.Utility; namespace Jinher.AMP.SNS.Chat.SocketManager { /// <summary> /// 消息服务基类 /// </summary> public abstract class MessageCenter { /// <summary> /// 表示是否连接上 /// </summary> public static bool IsConnected { get; protected set; } /// <summary> /// 表示是否进行第一次握手协议 /// </summary> public static bool IsHandStake { get; protected set; } /// <summary> /// 表示是否注册APP /// </summary> public static bool IsRegisterApp { get; protected set; } private static string _ip = string.Empty; ; private static object lockObject = new object(); static List<byte[]> byteList = new List<byte[]>(); //通知 private static SocketAsyncEventArgsPool pool = null; private static BufferManager m_bufferManager = null; /// <summary> /// 线程休眠时间(毫秒) /// </summary> ; //连接对象 private static Socket _socket = null; static MessageCenter() { //初始化,获取Host和Port _ip = System.Configuration.ConfigurationManager.AppSettings["serverip"]; _port = int.Parse(System.Configuration.ConfigurationManager.AppSettings["port"]); "; threadSleepTime = int.Parse(timer); //初始化连接对象 InitSocket(); } /// <summary> /// 启动消息中心 /// </summary> public static void Run() { try { LogHelper.WriteLog("启动服务"); //第一次连接服务器 Connect(); //保持连接 KeepConnect(); //推送消息队列 //SendAsync(); client.OnMsgReceiveed += client_OnMsgReceiveed; } catch (SocketException exception) { //socket出错 LogHelper.WriteLog("出错", exception); } catch (Exception ex) { //未知错误 } finally { //重置连接 //ResetConnect(); } } #region 连接服务器 /// <summary> /// 连接服务器 /// </summary> protected static async void Connect() { try { if (_socket == null) { InitSocket(); } LogHelper.WriteLog("开始建立连接"); _socket.Connect(IPAddress.Parse(_ip), _port); LogHelper.WriteLog("连接已经建立"); } catch (Exception ex) { LogHelper.WriteLog("出错了", ex); LogHelper.WriteLog("建立连接失败,等待重新建立连接"); Thread.Sleep(threadSleepTime); Connect(); return; } //先进行第一次握手协议 HandShakeCmdOp(); //注册app RegisterApp(); IsConnected = true; } /// <summary> /// 先进行第一次握手协议 /// </summary> private static void HandShakeCmdOp() { if (!IsHandStake) { LogHelper.WriteLog("开始第一次握手"); ); SendByInit(pmsMessage.HandShakePacket(handBytes)); IsHandStake = true; LogHelper.WriteLog("成功握手"); } } /// <summary> /// 注册app /// </summary> private static void RegisterApp() { // 一级命令:XNS_ROUTER //二级命令:REGISTER_SOCIAL_APP //并且规定AppId:99999999 LogHelper.WriteLog("开始注册APP"); string rapp = "<xns=XNS_ROUTER><cmd=REGISTER_SOCIAL_APP><appid=99999999>\0"; //rapp += "<clienttype=3><companyid=0><userid=0>\0"; byte[] bytes = Encoding.Default.GetBytes(rapp); _socket.Send(pmsMessage.RegisterPacket(bytes)); AccpetOne(); IsRegisterApp = true; LogHelper.WriteLog("注册成功"); } /// <summary> /// 初始化连接对象 /// </summary> private static void InitSocket() { _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); } #endregion #region 发送消息 /// <summary> /// 发送消息(异步) /// </summary> /// <param name="buffer"></param> /// <param name="count">表示失败重新发送次数</param> /// <returns></returns> ) { //先将该消息尝试发送,如果发送失败,则连接后继续发送 //byteList.Add(buffer); //启动任务,发送消息 Task.Factory.StartNew(new Action(() => { try { if (_socket != null && IsConnected && _socket.Connected) { lock (lockObject) { if (_socket != null && IsConnected && _socket.Connected) { SendByInit(buffer); } } } else { byteList.Add(buffer); } } catch (Exception) { //重发一次 ) { //表示失败重新发送 SendAsync(buffer, count + ); } } finally { } })); } /// <summary> /// 发送数据() /// </summary> /// <param name="buffer"></param> private static void SendByInit(byte[] buffer) { _socket.BeginSend(buffer, , buffer.Length, SocketFlags.None, new AsyncCallback(SendComplated), _socket); } /// <summary> /// 发送消息回调函数 /// </summary> /// <param name="ar"></param> private static void SendComplated(IAsyncResult async) { try { Socket skt = async.AsyncState as Socket; if (skt.Connected) { skt.EndSend(async); } LogHelper.WriteLog("发送成功"); } catch (SocketException ex) { //日志文件 } } #endregion #region 接受消息(方式一) /// <summary> /// 接受消息方式一 /// </summary> public static void AccpetOne() { LogHelper.WriteLog("开始建立接受消息(方式一)"); pool = ); m_bufferManager = * * , * ); m_bufferManager.InitBuffer(); // 预先分配一个对象池 SocketAsyncEventArgs readWriteEventArg; ; i < ; i++) { //初始化 SocketAsyncEventArgs readWriteEventArg = new SocketAsyncEventArgs(); readWriteEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(Receive_Completed); readWriteEventArg.UserToken = new AsyncUserToken(); // 从缓冲池分配一个字节缓冲区SocketAsyncEventArg对象 m_bufferManager.SetBuffer(readWriteEventArg); // add SocketAsyncEventArg to the pool pool.Push(readWriteEventArg); } //取出 监视 Receive(); } private static void Receive() { SocketAsyncEventArgs readEventArgs = null; readEventArgs = pool.Pop(); if (readEventArgs != null) { ((AsyncUserToken)readEventArgs.UserToken).Socket = _socket; _socket.ReceiveAsync(readEventArgs); } } private static void Receive_Completed(object sender, SocketAsyncEventArgs e) { ReceiveAsync(e); } private static void ReceiveAsync(SocketAsyncEventArgs e) { AsyncUserToken token = (AsyncUserToken)e.UserToken; && e.SocketError == SocketError.Success) { //echo the data received back to the client e.SetBuffer(, e.BytesTransferred); , e.BytesTransferred); //Console.WriteLine(tmp); //通知 #region 收到消息平台发送来的消息 //client.Send(); ) { ProcessAccpet(e.Buffer, e.BytesTransferred); } #endregion //回收 pool.Push(e); m_bufferManager.FreeBuffer(e); m_bufferManager.SetBuffer(e); Receive(); } } #endregion #region 接受消息(方式二) private static void AccpetTo() { LogHelper.WriteLog("开始建立接受消息(方式二)"); ; i < ; i++) { Task.Factory.StartNew(new Action(() => { try { * ]; while (true) { int r = _socket.Receive(bytes); //通知 #region 收到消息平台发送来的消息 //client.Send(); ) { //ClientSend((byte[])bytes.Clone(), r); ProcessAccpet((byte[])bytes.Clone(), r); } #endregion //string txt = Encoding.UTF8.GetString(bytes, 0, r); //Console.WriteLine(txt); bytes = * ]; } } catch (SocketException ex) { LogHelper.WriteLog("socket出错", ex); ResetConnect(); } catch (Exception ex) { LogHelper.WriteLog("出错", ex); } })); } } #endregion #region 收到的消息进行处理 /// <summary> /// 消息处理中转 /// </summary> /// <param name="p"></param> /// <param name="r"></param> private static void ProcessAccpet(byte[] p, int r) { //加入日志 //还没有进行握手,发送来的是握手包 if (IsHandStake) { //返回去握手命令 } ) { LogHelper.WriteLog("收到正常消息"); var result = pmsMessage.ReversePacketMessage(p, r); } else { LogHelper.WriteLog(string.Format("收到心跳包")); } } #endregion #region 保持连接 /// <summary> /// 重置连接 /// </summary> private static void ResetConnect() { LogHelper.WriteLog("=========重新连接=============="); lock (lockObject) { IsConnected = false; _socket = null; InitSocket(); } Connect(); } private static Timer timer = null; /// <summary> /// 保持连接 /// </summary> private static void KeepConnect() { timer = new Timer(new TimerCallback((o) => { //发送心跳包 SendAsync(pmsMessage.GetHeardbeat()); LogHelper.WriteLog("已经发送心跳包"); }), , , , ), , , , )); } #endregion #region 通知相关 private static ChatClient client = ChatClient.Instance(ConfigurationManager.AppSettings["notificationUri"]); static PublishWebMessage pmsMessage = new PublishWebMessage(); /// <summary> /// 请求发送消息 /// </summary> /// <param name="msg"></param> protected static void client_OnMsgReceiveed(WebMessageDTO msg) { byte[] bytes = pmsMessage.PacketMessage(msg); SendAsync(bytes); } private static void ClientSend(byte[] buffer, int length) { MessageBase message = pmsMessage.ReversePacketMessage(buffer, length); } #endregion } }
最新文章
- 【分布式】Zookeeper使用--开源客户端
- SQL 分页
- HA模式下历史服务器配置
- Swift3.0语言教程查找字符集和子字符串
- Node.js的高性能封装 Express.js
- jQuery插件实现图片展开效果,jquery.gallery。仿腾讯QQ空间说说图片展示效果。
- Sqli-labs less 44
- 书写CSS需要注意的地方
- SET ANSI_NULLS ON
- BZOJ 1048 分割矩阵
- 阿里云服务器[教程3]一键安装php+mysql+ftp+nginx环境
- Hadoop HDFS (3) JAVA訪问HDFS之二 文件分布式读写策略
- 其他主机连接本地主机Tomcat会出现的防火墙问题
- Python学习笔记——基础篇【第二周】——解释器、字符串、列表、字典、主文件判断、对象
- javacpp-opencv图像处理系列:国内车辆牌照检测识别系统(万份测试车牌识别准确率99.7%以上,单次平均耗时39ms)
- Java 垃圾回收算法
- tmux 使用说明
- Apollo 7 — ConfigService 消息扫描设计实现
- REST-framework快速构建API--分页
- Java8 使用