实时通讯第二篇

namespace FastDev.IocpSocket
{
    public class SocketEventPool
    {
        Stack<SocketAsyncEventArgs> m_pool;


        public SocketEventPool(int capacity)
        {
            m_pool = new Stack<SocketAsyncEventArgs>(capacity);
        }

        public void Push(SocketAsyncEventArgs item)
        {
            if (item == null) { throw new ArgumentNullException("Items added to a SocketAsyncEventArgsPool cannot be null"); }
            lock (m_pool)
            {
                m_pool.Push(item);
            }
        }

        //从池中移除 SocketAsyncEventArgs 实例 
        // 并返回从池中移除的对象  
        public SocketAsyncEventArgs Pop()
        {
            lock (m_pool)
            {
                return m_pool.Pop();
            }
        }

        //池中的 SocketAsyncEventArgs 实例的数目 
        public int Count
        {
            get { return m_pool.Count; }
        }

        public void Clear()
        {
            m_pool.Clear();
        }
    }
}

namespace FastDev.IocpSocket
{
    public class AsyncUserToken
    {
        /// <summary>  
        /// 客户端IP地址  
        /// </summary>  
        public IPAddress IPAddress { get; set; }

        /// <summary>  
        /// 远程地址  
        /// </summary>  
        public EndPoint Remote { get; set; }

        /// <summary>  
        /// 通信SOKET  
        /// </summary>  
        public Socket Socket { get; set; }

        /// <summary>  
        /// 连接时间  
        /// </summary>  
        public DateTime ConnectTime { get; set; }

        ///// <summary>  
        ///// 所属用户信息  
        ///// </summary>  
        //public UserInfoModel UserInfo { get; set; }
        /// <summary>  
        /// 咱这里就储存一个CompanyId
        /// </summary>  
        public ReceiveMsgModel Model { get; set; }

        /// <summary>  
        /// 数据缓存区  
        /// </summary>  
        public List<byte> Buffer { get; set; }


        public AsyncUserToken()
        {
            this.Buffer = new List<byte>();
        }
    }
}

namespace FastDev.IocpSocket
{
    public static class SocketManager
    {

        private static int m_maxConnectNum;    //最大连接数  
        private static int m_revBufferSize;    //最大接收字节数  
        static BufferManager m_bufferManager;
        const int opsToAlloc = 2;
        static Socket listenSocket;            //监听Socket  
        static SocketEventPool m_pool;
        static int m_clientCount;              //连接的客户端数量  
        static Semaphore m_maxNumberAcceptedClients;

        static List<AsyncUserToken> m_clients; //客户端列表  

        #region 定义委托  

        /// <summary>  
        /// 客户端连接数量变化时触发  
        /// </summary>  
        /// <param name="num">当前增加客户的个数(用户退出时为负数,增加时为正数,一般为1)</param>  
        /// <param name="token">增加用户的信息</param>  
        //public delegate void OnClientNumberChange(int num, AsyncUserToken token);

        /// <summary>  
        /// 接收到客户端的数据  
        /// </summary>  
        /// <param name="token">客户端</param>  
        /// <param name="buff">客户端数据</param>  
        public delegate void OnReceiveData(AsyncUserToken token, byte[] buff);

        #endregion

        #region 定义事件  
        /// <summary>  
        /// 客户端连接数量变化事件  
        /// </summary>  
        //public OnClientNumberChange ClientNumberChange;

        /// <summary>  
        /// 接收到客户端的数据事件  
        /// </summary>  
        public static OnReceiveData ReceiveClientData;
        #endregion

        #region 定义属性  

        /// <summary>  
        /// 获取客户端列表  
        /// </summary>  
        public static List<AsyncUserToken> ClientList { get { return m_clients; } }

        #endregion

        /// <summary>  
        /// 构造函数  
        /// </summary>  
        /// <param name="numConnections">最大连接数</param>  
        /// <param name="receiveBufferSize">缓存区大小</param>  
        static SocketManager()
        {
            m_clientCount = 0;
            m_maxConnectNum = 200;
            m_revBufferSize = 1200;
            // 分配缓冲区，以便最大数量的套接字可以有一个未读数，同时向 Socket 写入    
            m_bufferManager = new BufferManager(1200 * 200 * opsToAlloc, 1200);

            m_pool = new SocketEventPool(200);
            m_maxNumberAcceptedClients = new Semaphore(200, 200);
        }

        /// <summary>  
        /// 初始化  
        /// </summary>  
        public static void Init()
        {
            // 分配一个大字节缓冲区，所有I/O操作都使用一个。这个防止内存碎片化
            m_bufferManager.InitBuffer();
            m_clients = new List<AsyncUserToken>();
            //SocketAsyncEventArgs 对象的预分配池
            SocketAsyncEventArgs readWriteEventArg;

            for (int i = 0; i < m_maxConnectNum; i++)
            {
                readWriteEventArg = new SocketAsyncEventArgs();
                readWriteEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
                readWriteEventArg.UserToken = new AsyncUserToken();

                // 将缓冲区中的字节缓冲区分配给 SocketAsyncEventArg 对象  
                m_bufferManager.SetBuffer(readWriteEventArg);
                // 向池中添加 SocketAsyncEventArg
                m_pool.Push(readWriteEventArg);
            }
        }


        /// <summary>  
        /// 启动服务  
        /// </summary>  
        /// <param name="localEndPoint"></param>  
        public static bool Start(IPEndPoint localEndPoint)
        {
            try
            {
                m_clients.Clear();
                listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
                listenSocket.Bind(localEndPoint);
                // 使用100个连接的侦听后台日志启动服务器  
                listenSocket.Listen(m_maxConnectNum);
                // post accepts on the listening socket  
                StartAccept(null);
                return true;
            }
            catch (Exception)
            {
                return false;
            }
        }

        /// <summary>  
        /// 停止服务  
        /// </summary>  
        public static void Stop()
        {
            foreach (AsyncUserToken token in m_clients)
            {
                try
                {
                    token.Socket.Shutdown(SocketShutdown.Both);
                }
                catch (Exception) { }
            }
            try
            {
                listenSocket.Shutdown(SocketShutdown.Both);
            }
            catch (Exception) { }

            listenSocket.Close();
            int c_count = m_clients.Count;
            lock (m_clients) { m_clients.Clear(); }

            //ClientNumberChange?.Invoke(-c_count, null);
        }


        public static void CloseClient(AsyncUserToken token)
        {
            try
            {
                token.Socket.Shutdown(SocketShutdown.Both);
            }
            catch (Exception) { }
        }


        // 开始接受客户端的连接请求的操作   
        //  
        // <param name="acceptEventArg">在服务器侦听套接字上发出接受操作时要使用的上下文对象</param>  
        public static void StartAccept(SocketAsyncEventArgs acceptEventArg)
        {
            if (acceptEventArg == null)
            {
                acceptEventArg = new SocketAsyncEventArgs();
                acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);
            }
            else
            {
                // 由于上下文对象正在重用，必须清除套接字。  
                acceptEventArg.AcceptSocket = null;
            }

            m_maxNumberAcceptedClients.WaitOne();
            if (!listenSocket.AcceptAsync(acceptEventArg))
            {
                ProcessAccept(acceptEventArg);
            }
        }

        // 此方法是与SoCK.Access同步的回调方法操作，并在接受操作完成时调用  

        static void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs e)
        {
            ProcessAccept(e);
        }

        private static void ProcessAccept(SocketAsyncEventArgs e)
        {
            try
            {
                Interlocked.Increment(ref m_clientCount);
                // 获取接受的客户端连接的套接字，并将其放入对象用户令牌
                SocketAsyncEventArgs readEventArgs = m_pool.Pop();
                AsyncUserToken userToken = (AsyncUserToken)readEventArgs.UserToken;
                userToken.Socket = e.AcceptSocket;
                userToken.ConnectTime = DateTime.Now;
                userToken.Remote = e.AcceptSocket.RemoteEndPoint;

                userToken.IPAddress = ((IPEndPoint)(e.AcceptSocket.RemoteEndPoint)).Address;

                lock (m_clients) { m_clients.Add(userToken); }

                //ClientNumberChange?.Invoke(1, userToken);
                if (!e.AcceptSocket.ReceiveAsync(readEventArgs))
                {
                    ProcessReceive(readEventArgs);
                }
            }
            catch (Exception me)
            {

            }

            // Accept the next connection request  
            if (e.SocketError == SocketError.OperationAborted) return;
            StartAccept(e);
        }


        static void IO_Completed(object sender, SocketAsyncEventArgs e)
        {
            // 确定刚刚完成的操作类型并调用关联的处理程序  
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Receive:
                    ProcessReceive(e);
                    break;
                case SocketAsyncOperation.Send:
                    ProcessSend(e);
                    break;
                default:
                    throw new ArgumentException("The last operation completed on the socket was not a receive or send");
            }

        }


        // 当异步接收操作完成时调用此方法.   
        // 如果远程主机关闭连接，则套接字关闭.    
        // 如果接收到数据，则将数据回传给客户端.  
        //  
        private static void ProcessReceive(SocketAsyncEventArgs e)
        {
            try
            {
                // 检查远程主机是否关闭了连接  
                AsyncUserToken token = (AsyncUserToken)e.UserToken;
                if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
                {
                    //读取数据  
                    byte[] data = new byte[e.BytesTransferred];
                    Array.Copy(e.Buffer, e.Offset, data, 0, e.BytesTransferred);
                    lock (token.Buffer)
                    {
                        token.Buffer.AddRange(data);
                    }
                    //注意:你一定会问,这里为什么要用do-while循环?   
                    //如果当客户发送大数据流的时候,e.BytesTransferred的大小就会比客户端发送过来的要小,  
                    //需要分多次接收.所以收到包的时候,先判断包头的大小.够一个完整的包再处理.  
                    //如果客户短时间内发送多个小数据包时, 服务器可能会一次性把他们全收了.  
                    //这样如果没有一个循环来控制,那么只会处理第一个包,  
                    //剩下的包全部留在token.Buffer中了,只有等下一个数据包过来后,才会放出一个来.  
                    do
                    {
                        //判断包的长度  
                        byte[] lenBytes = token.Buffer.GetRange(0, 4).ToArray();
                        int packageLen = BitConverter.ToInt32(lenBytes, 0);
                        if (packageLen > token.Buffer.Count - 4)
                        {   //长度不够时,退出循环,让程序继续接收  
                            break;
                        }

                        //包够长时,则提取出来,交给后面的程序去处理  
                        byte[] rev = token.Buffer.GetRange(4, packageLen).ToArray();
                        //从数据池中移除这组数据  
                        lock (token.Buffer)
                        {
                            token.Buffer.RemoveRange(0, packageLen + 4);
                        }
                        //这里回调--各自的方法 --我这里做数据库处理
                        ReceiveClientData = new OnReceiveData(ReceiveMsg);
                        //将数据包交给后台处理,这里你也可以新开个线程来处理.加快速度.  
                        ReceiveClientData.Invoke(token, rev);
                        //这里API处理完后,并没有返回结果,当然结果是要返回的,却不是在这里, 这里的代码只管接收.  
                        //若要返回结果,可在API处理中调用此类对象的SendMessage方法,统一打包发送.不要被微软的示例给迷惑了.  
                    } while (token.Buffer.Count > 4);

                    //继续接收. 为什么要这么写,请看Socket.ReceiveAsync方法的说明  
                    if (!token.Socket.ReceiveAsync(e))
                        ProcessReceive(e);
                }
                else
                {
                    CloseClientSocket(e);
                }
            }
            catch (Exception xe)
            {

            }
        }
        /// <summary>
        /// 处理客户端消息
        /// </summary>
        /// <param name="token"></param>
        /// <param name="b"></param>
        private static void ReceiveMsg(AsyncUserToken token, byte[] b)
        {
            string msg = System.Text.Encoding.Default.GetString(b);
            if (!string.IsNullOrWhiteSpace(msg))
            {
                ReceiveMsgModel model = TypeConvertHepler.JsonToData<ReceiveMsgModel>(msg);
                if (model != null && model.State != 0)
                {
                    switch (model.State)
                    {
                        case 1:
                            foreach (var temp in m_clients)
                            {
                                if (temp.Remote == token.Remote)//通讯地址相等
                                {

                                    temp.Model = new ReceiveMsgModel
                                    {
                                        CompanyId = model.CompanyId
                                    };
                                }
                            }
                            break;
                        case 2:
                            //操作数据库
                            Go.Travel.NetWork.WcfHelper.TravelHelper.UpdateByWhere_B2cmesremind(new Dictionary<string, object> {
                                { nameof(B2cmesremind.IsRead),1}
                            }, new Dictionary<string, object> {
                                { nameof(B2cmesremind.KID),model.KID }
                            });
                            break;
                    }

                }
            }
        }
        // 异步发送操作完成时调用的方法.    
        // 该方法在套接字上发出另一个接收来读取任何附加信息   
        // 客户端发送的数据  
        //  
        // <param name="e"></param>  
        private static void ProcessSend(SocketAsyncEventArgs e)
        {
            if (e.SocketError == SocketError.Success)
            {
                // 完成对客户端的数据回传  
                AsyncUserToken token = (AsyncUserToken)e.UserToken;
                // 读取客户端发送的下一个数据块  
                bool willRaiseEvent = token.Socket.ReceiveAsync(e);
                if (!willRaiseEvent)
                {
                    ProcessReceive(e);
                }
            }
            else
            {
                CloseClientSocket(e);
            }
        }

        //关闭客户端  
        private static void CloseClientSocket(SocketAsyncEventArgs e)
        {
            AsyncUserToken token = e.UserToken as AsyncUserToken;

            lock (m_clients) { m_clients.Remove(token); }
            //如果有事件,则调用事件,发送客户端数量变化通知  
            //ClientNumberChange?.Invoke(-1, token);
            // 关闭与客户端关联的套接字  
            try
            {
                token.Socket.Shutdown(SocketShutdown.Send);
            }
            catch (Exception) { }
            token.Socket.Close();
            // 减少计数器，以跟踪连接到服务器的客户端总数  
            Interlocked.Decrement(ref m_clientCount);
            m_maxNumberAcceptedClients.Release();
            //SocketAsyncEventArg，以便它们可以被另一个客户端重用。
            e.UserToken = new AsyncUserToken();
            m_pool.Push(e);
        }



        /// <summary>  
        /// 对数据进行打包,然后再发送  
        /// </summary>  
        /// <param name="token"></param>  
        /// <param name="message"></param>  
        /// <returns></returns>  
        public static void SendMessage(AsyncUserToken token, byte[] message)
        {
            if (token == null || token.Socket == null || !token.Socket.Connected)
                return;
            try
            {
                //对要发送的消息,制定简单协议,头4字节指定包的大小,方便客户端接收(协议可以自己定)  
                byte[] buff = new byte[message.Length + 4];
                byte[] len = BitConverter.GetBytes(message.Length);
                Array.Copy(len, buff, 4);
                Array.Copy(message, 0, buff, 4, message.Length);
                //token.Socket.Send(buff);  //这句也可以发送, 可根据自己的需要来选择  
                //新建异步发送对象, 发送消息  
                SocketAsyncEventArgs sendArg = new SocketAsyncEventArgs();
                sendArg.UserToken = token;
                sendArg.SetBuffer(buff, 0, buff.Length);  //将数据放置进去.  
                token.Socket.SendAsync(sendArg);
            }
            catch (Exception e)
            {

            }
        }

        /// <summary>
        /// 给所有在线的用户发送消息,且CompanyId == 传递的值
        /// </summary>
        /// <param name="message"></param>
        /// <param name="companyId"></param>
        public static void SendAllUserMsg(byte[] message, string companyId)
        {
            try
            {
                foreach (var token in m_clients)
                {
                    if (token == null || token.Socket == null || !token.Socket.Connected || token.Model.CompanyId != companyId)
                        continue;
                    else
                    {
                        //对要发送的消息,制定简单协议,头4字节指定包的大小,方便客户端接收(协议可以自己定)  
                        byte[] buff = new byte[message.Length + 4];
                        byte[] len = BitConverter.GetBytes(message.Length);
                        Array.Copy(len, buff, 4);
                        Array.Copy(message, 0, buff, 4, message.Length);
                        //token.Socket.Send(buff);  //这句也可以发送, 可根据自己的需要来选择  
                        //新建异步发送对象, 发送消息  
                        SocketAsyncEventArgs sendArg = new SocketAsyncEventArgs();
                        sendArg.UserToken = token;
                        sendArg.SetBuffer(buff, 0, buff.Length);  //将数据放置进去.  
                        token.Socket.SendAsync(sendArg);
                    }
                }
                //调用http推送消息  System.Text.Encoding.Default.GetString(message)
                SendMsgModel model = TypeConvertHepler.JsonToData<SendMsgModel>(Encoding.Default.GetString(message));
                if (model != null)
                {
                    HttpHepler.HttpGet(ConfigurationManager.AppSettings["lyUrlPath"].ToString() + "/SignalrPush/PushMsg/?message=" + model.Guid);
                }
            }
            catch (Exception e)
            {

            }
        }
    }
}