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 int _port = 0;
        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 int threadSleepTime = 5000;
        //连接对象
        private static Socket _socket = null;

        static MessageCenter()
        {
            //初始化，获取Host和Port
            _ip = System.Configuration.ConfigurationManager.AppSettings["serverip"];
            _port = int.Parse(System.Configuration.ConfigurationManager.AppSettings["port"]);

            string timer = System.Configuration.ConfigurationManager.AppSettings["threadSleepTime"] ?? "5000";
            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("开始第一次握手");
                byte[] handBytes = BitConverter.GetBytes(0);
                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>
        protected static void SendAsync(byte[] buffer, int count = 0)
        {
            //先将该消息尝试发送，如果发送失败，则连接后继续发送
            //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)
                {
                    //重发一次
                    if (count == 0)
                    {
                        //表示失败重新发送
                        SendAsync(buffer, count + 1);
                    }
                }
                finally
                {

                }

            }));

        }
        /// <summary>
        /// 发送数据()
        /// </summary>
        /// <param name="buffer"></param>
        private static void SendByInit(byte[] buffer)
        {
            _socket.BeginSend(buffer, 0, 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 = new SocketAsyncEventArgsPool(10);
            m_bufferManager = new BufferManager(1024 * 1024 * 10, 1024 * 1024);
            m_bufferManager.InitBuffer();
            // 预先分配一个对象池
            SocketAsyncEventArgs readWriteEventArg;

            for (int i = 0; i < 10; 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;
            if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
            {
                //echo the data received back to the client
                e.SetBuffer(0, e.BytesTransferred);
                string txt = Encoding.UTF8.GetString(e.Buffer, 0, e.BytesTransferred);
                //Console.WriteLine(tmp);
                //通知
                #region 收到消息平台发送来的消息
                //client.Send();
                if (e.BytesTransferred > 0)
                {
                    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("开始建立接受消息（方式二）");
            for (int i = 0; i < 1; i++)
            {
                Task.Factory.StartNew(new Action(() =>
                {
                    try
                    {
                        byte[] bytes = new byte[1024 * 1024];
                        while (true)
                        {
                            int r = _socket.Receive(bytes);
                            //通知

                            #region 收到消息平台发送来的消息

                            //client.Send();
                            if (r > 0)
                            {
                                //ClientSend((byte[])bytes.Clone(), r);
                                ProcessAccpet((byte[])bytes.Clone(), r);

                            }

                            #endregion

                            //string txt = Encoding.UTF8.GetString(bytes, 0, r);
                            //Console.WriteLine(txt);
                            bytes = new byte[1024 * 1024];
                        }
                    }
                    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)
            {
                //返回去握手命令
            }
            if (r > 8)
            {
                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("已经发送心跳包");
            }), null, new TimeSpan(0, 0, 0, 5), new TimeSpan(0, 0, 0, 10));

        }

        #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

    }
}