C# socket 实现消息中心向消息平台 转发消息

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 }}