C＃在每个线程中建立一个连接

Question

给你参考一下我的代码
namespace GPSSubGateWayServer
{
    class GPSCarService
    {
        public event ReceiveClientUDPMessageEventHandle OnReceiveClientUDPMessageEvent;
        public event ReceiveClientTCPMessageEventHandle OnReceiveClientTCPMessageEvent;
        public event ClientLoginEventHandle OnClientLoginEvent;
        public event ClientLogOutEventHandle OnClientLogOutEvent;
        Socket ServerSock;
        IPAddress ServerIP;
        int Port;
        /// <summary>
        /// 记录所有登陆过的ClientName,停止时使用
        /// </summary>
        ArrayList al_ClientName = ArrayList.Synchronized(new ArrayList());
        ArrayList al_SendMessageOutQueue = ArrayList.Synchronized(new ArrayList());
        /// <summary>
        /// 存储车辆名称与接收线程
        /// </summary>
        //SortedList sl_Thread_ReceiveThread = SortedList.Synchronized(new SortedList());
        /// <summary>
        /// 存储车辆名称与接收信息的Socket;
        /// </summary>
        //SortedList sl_Socket_ClientName = SortedList.Synchronized(new SortedList());
        /// <summary>
        /// 　存储车辆名称与Socket保护锁
        /// </summary>
        //SortedList sl_ClientSocketReaderWriterLock_ClientName = SortedList.Synchronized(new SortedList());
        SortedList sl_ClientState_ClientName = SortedList.Synchronized(new SortedList());
        ReaderWriterLock RWL_sl_ClientState_ClientName = new ReaderWriterLock();
        Thread ListenClientThread;
        ArrayList al_ClientEventArgs = ArrayList.Synchronized(new ArrayList());
        Thread MoniteClintThread;
        /// <summary>
        /// 标志是否允许启动
        /// </summary>
        bool IFStart = false;
        AutoResetEvent ARE_Listen = new AutoResetEvent(true);
        //Thread TH_ReceiveNewClientLoginMessage;
        // Thread TH_ReceiveNewClientLoginMessageMonite;
        Thread TH_SendEvent;
        Thread TH_SendMessage;
        Thread TH_RecvUDPMessage;
        System.Net.Sockets.UdpClient UdpReceiver;
        public void SendMessage(MessageSendQueueItem Item)
        {
            this.al_SendMessageOutQueue.Add(Item);
        }
        public GPSCarService(string ServerIP, int ServerPort)
        {
            this.ServerSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            this.ServerIP = IPAddress.Parse(ServerIP);
            this.Port = ServerPort;
            this.UdpReceiver = new UdpClient(this.Port);
        }
        public void Start()
        {
            IPEndPoint ipep = new IPEndPoint(this.ServerIP, this.Port);
            this.ServerSock.Bind(ipep);
            this.ServerSock.Listen(10);
            this.IFStart = true;
            //启动监控线程
            this.MoniteClintThread = new Thread(new ThreadStart(this.ThreadMoniteClient));
            this.MoniteClintThread.Start();
            //启动发送事件线程
            this.TH_SendEvent = new Thread(new ThreadStart(this.ThreadEvent));
            this.TH_SendEvent.Start();
            //启动发送信息线程
            this.TH_SendMessage = new Thread(new ThreadStart(this.ThreadSendMessage));
            this.TH_SendMessage.Start();
            //启动接收UDP的线程
            this.TH_RecvUDPMessage = new Thread(new ThreadStart(this.ReceiveClientUDPMessage));
            this.TH_RecvUDPMessage.Start();
            //启动侦听连接线程
            this.ListenClientThread = new Thread(new ThreadStart(this.ThreadListen));
            this.ListenClientThread.Start();
        }     
        private void ThreadEvent()
        {
            while (this.IFStart)
            {
                Thread.Sleep(1);
                if (this.al_ClientEventArgs.Count > 0)
                {
                    EventArgsItem eai = (EventArgsItem)this.al_ClientEventArgs[0];
                    this.al_ClientEventArgs.RemoveAt(0);
                    switch (eai.Type)
                    {
                        case EventArgsItemType.ClientLoginEventArgs:
                            {
                                ClientLoginEventArgs ea = (ClientLoginEventArgs)eai.Item;
                                //this.Writefile(DateTime.Now.ToString() + "\tClient " + ea.Name + " 连接");
                                if (this.OnClientLoginEvent != null)
                                {
                                    this.OnClientLoginEvent(ea, this);
                                }
                            }
                            break;
                        case EventArgsItemType.ReceiveClientTCPMessageEventArgs:
                            {
                                ReceiveClientTCPMessageEventArgs ea = (ReceiveClientTCPMessageEventArgs)eai.Item;
                                if (this.OnReceiveClientTCPMessageEvent != null)
                                {
                                    this.OnReceiveClientTCPMessageEvent(ea, this);
                                }
                                //this.Writefile(ea.ReceiveTime.ToString() + "\t" + ea.Name + "\t" + "TCP\t" + Convert.ToBase64String(ea.Message));                            }
                            break;
                        case EventArgsItemType.ReceiveClientUDPMessageEventArgs:
                            {
                                ReceiveClientUDPMessageEventArgs ea = (ReceiveClientUDPMessageEventArgs)eai.Item;
                                if (this.OnReceiveClientUDPMessageEvent != null)
                                {
                                    this.OnReceiveClientUDPMessageEvent(ea, this);
                                }
                                //this.Writefile(ea.ReceiveTime.ToString() + "\t" + ea.Name + "\t" + "UDP\t" + Convert.ToBase64String(ea.Message));
                            }
                            break;
                        case EventArgsItemType.ClientLogOutEventArgs:
                            {
                                ClientLogOutEventArgs ea = (ClientLogOutEventArgs)eai.Item;
                                if (this.OnClientLogOutEvent != null)
                                {
                                    this.OnClientLogOutEvent(ea, this);
                                }
                                //this.Writefile(DateTime.Now.ToString() + "\tClient " + ea.ClientName + " 断开");
                            }
                            break;
                    }
                }
            }
        }
        private void ThreadListen()
        {
            while (this.IFStart)
            {
                this.ARE_Listen.Reset();
                this.ServerSock.BeginAccept(new AsyncCallback(this.AcceptCallBack), null);
                this.ARE_Listen.WaitOne();
                Thread.Sleep(1000);
            }        }
        private void WriteLog(string LogMessage)
        {
            lock (this)
            {
                if (!System.IO.Directory.Exists(System.Windows.Forms.Application.StartupPath + "\\LOG\\GPSServiceLog"))
                {
                    System.IO.Directory.CreateDirectory(System.Windows.Forms.Application.StartupPath + "\\LOG\\GPSServiceLog");
                }
                string filepath = System.Windows.Forms.Application.StartupPath + "\\LOG\\GPSServiceLog\\" + DateTime.Now.ToLongDateString() + ".txt";
                System.IO.FileStream fs = new FileStream(filepath, FileMode.Append, FileAccess.Write, FileShare.Read);
                StreamWriter sw = new StreamWriter(fs, System.Text.Encoding.UTF8);
                sw.WriteLine(DateTime.Now.ToLongTimeString() + "\t" + LogMessage);
                sw.Flush();
                fs.Flush();
                sw.Close();
                fs.Close();
            }
        }
}

Answer 1

 private void AsyncReceiveClientTCPMessge(System.IAsyncResult ar)
        {
            ClientState state = (ClientState)ar.AsyncState;            int size = 0;
            if (state == null)
            {
                return;
            }
            if (state.StateFlag > 1)
            {
                //资源已经被清理，不需要重复清理
                return;
            }
            state.RWL_ClientSock.AcquireWriterLock(-1);
            try
            {
                size = state.ClientSocket.EndReceive(ar);
            }
            catch (Exception ex)
            {
                //接收失败，清理资源
                string str = ex.Message;
                string log = "接收信息异常,原因-->" + ex.Message + "ClientName:" + state.ClientName;
                this.WriteLog(log);
                //先释放锁
                state.RWL_ClientSock.ReleaseLock();
                //this.sl_ClientState_ClientName.Remove(state.ClientName);
                //申请全局保护锁
                this.RWL_sl_ClientState_ClientName.AcquireWriterLock(-1);
                if (this.sl_ClientState_ClientName.ContainsKey(state.ClientName))
                {
                    //清除全局变量
                    //ClientState _cs = (ClientState)this.sl_ClientState_ClientName[state.ClientName];
                    if (state.StateFlag == 1)
                    {                        //说明是车机断开，此处清理资源
                        System.Threading.Interlocked.Increment(ref state.StateFlag);
                        this.sl_ClientState_ClientName.Remove(state.ClientName);
                        try
                        {
                            //关闭连接
                            state.ClientSocket.Close();
                        }
                        catch
                        { }
                        EventArgsItem eia = new EventArgsItem();
                        eia.Type = EventArgsItemType.ClientLogOutEventArgs;
                        ClientLogOutEventArgs eea = new ClientLogOutEventArgs();
                        eea.LogoutTime = DateTime.Now;
                        eea.ClientName = state.ClientName;
                        eia.Item = eea;
                        this.al_ClientEventArgs.Add(eia);
                    }
                    //否则，是重复连接或连接已断                }
                this.RWL_sl_ClientState_ClientName.ReleaseLock();
                return;
            }
            if (size == 0)
            { //接收失败，清理资源
                //string str = ex.Message;
                string log = "接收信息异常,原因-->接收到Client发送来的信息为0字节.ClientName:" + state.ClientName;
                this.WriteLog(log);
                state.RWL_ClientSock.ReleaseLock();
                //this.sl_ClientState_ClientName.Remove(state.ClientName);
                this.RWL_sl_ClientState_ClientName.AcquireWriterLock(-1);
                if (this.sl_ClientState_ClientName.ContainsKey(state.ClientName))
                {
                    //ClientState _cs = (ClientState)this.sl_ClientState_ClientName[state.ClientName];
                    if (state.StateFlag == 1)
                    {
                        System.Threading.Interlocked.Increment(ref state.StateFlag);
                        this.sl_ClientState_ClientName.Remove(state.ClientName);
                        try
                        {
                            //关闭连接
                            state.ClientSocket.Close();
                        }
                        catch
                        { }
                        EventArgsItem eia = new EventArgsItem();
                        eia.Type = EventArgsItemType.ClientLogOutEventArgs;
                        ClientLogOutEventArgs eea = new ClientLogOutEventArgs();
                        eea.LogoutTime = DateTime.Now;
                        eea.ClientName = state.ClientName;
                        eia.Item = eea;
                        this.al_ClientEventArgs.Add(eia);
                    }
                }
                this.RWL_sl_ClientState_ClientName.ReleaseLock();
                return;
            }
            state.SetLastRecvClientMessageTime();
            state.RWL_ClientSock.ReleaseLock();
            EventArgsItem eai = new EventArgsItem();
            eai.Type = EventArgsItemType.ReceiveClientTCPMessageEventArgs;
            ReceiveClientTCPMessageEventArgs ea = new ReceiveClientTCPMessageEventArgs();
            ea.Message = new byte[size];
            Array.Copy(state.bsBuffer, 0, ea.Message, 0, size);
            ea.ReceiveTime = DateTime.Now;
            ea.Name = state.ClientName;
            eai.Item = ea;
            this.al_ClientEventArgs.Add(eai);
            state.bsBuffer.Initialize();// = new byte[1024];
            state.RWL_ClientSock.AcquireWriterLock(-1);
            try
            {
                state.ClientSocket.BeginReceive(state.bsBuffer, 0, 1024, SocketFlags.None, new AsyncCallback(this.AsyncReceiveClientTCPMessge), state);
            }
            catch (Exception ex)
            {
                //接收失败，清理资源
                string str = ex.Message;
                string log = "接收信息异常,原因-->" + ex.Message + "ClientName:" + state.ClientName;
                this.WriteLog(log);
                state.RWL_ClientSock.ReleaseLock();
                //this.sl_ClientState_ClientName.Remove(state.ClientName);
                this.RWL_sl_ClientState_ClientName.AcquireWriterLock(-1);
                if (this.sl_ClientState_ClientName.ContainsKey(state.ClientName))
                {
                    if (state.StateFlag == 1)
                    {
                        System.Threading.Interlocked.Increment(ref state.StateFlag);
                        this.sl_ClientState_ClientName.Remove(state.ClientName);
                        try
                        {
                            //关闭连接
                            state.ClientSocket.Close();
                        }
                        catch
                        { }
                        EventArgsItem eia = new EventArgsItem();
                        eia.Type = EventArgsItemType.ClientLogOutEventArgs;
                        ClientLogOutEventArgs eea = new ClientLogOutEventArgs();
                        eea.LogoutTime = DateTime.Now;
                        eea.ClientName = state.ClientName;
                        eia.Item = eea;
                        this.al_ClientEventArgs.Add(eia);
                    }                }
                this.RWL_sl_ClientState_ClientName.ReleaseLock();
                return;
            }
            state.RWL_ClientSock.ReleaseLock();        }
    }

Answer 2

private void AcceptCallBack(System.IAsyncResult ar)
        {
            Socket sock;
            try
            {
                sock = this.ServerSock.EndAccept(ar);
            }
            catch
            {
                //接受连接失败
                this.ARE_Listen.Set();
                return;
            }
            IPEndPoint ClientIPEP = (IPEndPoint)sock.RemoteEndPoint;
            this.WriteLog("有新连接接入,IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString());
            //this.NewSocke = sock;            //TH_ReceiveNewClientLoginMessage = new Thread(new ThreadStart(this.ReceiveNewClientLoginMessage));
            //TH_ReceiveNewClientLoginMessage.Start();
            //TH_ReceiveNewClientLoginMessageMonite = new Thread(new ThreadStart(this.ReceiveNewClientLoginMessageMonite));
            //TH_ReceiveNewClientLoginMessageMonite.Start();
            byte[] bsmsg = new byte[1024];
            int size = 0;
            try
            {
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString()+"\t准备连接");
                Socket tempSocket = sock;
                tempSocket.ReceiveTimeout = 1;
                size = tempSocket.Receive(bsmsg);
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString()+"\t接到连接后第一条信息");
            }
            catch(Exception ex)
            { 
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "连接成功后没有发送任何信息,准备关闭"+ex.Message);
                try
                {
                    sock.Close();
                }
                catch
                { }
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "连接被关闭");
                this.ARE_Listen.Set();
                return;
            }            if (size == 0)
            {
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "信息长度为0");
                try
                {
                    sock.Close();
                }
                catch
                { }
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "连接被关闭");
                this.ARE_Listen.Set();
                return;
            }
            string msg = "";
            try
            {
                msg = System.Text.Encoding.ASCII.GetString(bsmsg, 0, size);
            }
            catch
            {
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() +"转换字符时出现错误\t"+msg);
                try
                {
                    sock.Close();
                }
                catch
                { }
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "连接被关闭");
                this.ARE_Listen.Set();
                return;
            }
            // bsmsg.Initialize();
            // size=this.NewSocke.Receive(bsmsg,0,1024,System.Net.Sockets.SocketFlags.None);
            if (msg.Length < 14)
            {
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "消息长度不够\t"+msg);
                try
                {
                    sock.Close();
                }
                catch
                { }
                //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "连接被关闭");
                this.ARE_Listen.Set();
                return;
            }
            int i = 0;
            while (i < 10)
            {
                //检查接收数据的4-14位是否为数字，否则为错误数据
                if (!char.IsLetterOrDigit(msg[i + 4]))
                {
                    //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "无法识别的主机号\t"+msg);
                    try
                    {
                        sock.Close();
                    }
                    catch
                    { }
                    //this.WriteTemp("IP:" + ClientIPEP.Address.ToString() + "端口:" + ClientIPEP.Port.ToString() + "连接被关闭");
                    this.ARE_Listen.Set();
                    return;
                }
                i++;
            }
            string ClientName = msg.Substring(4, 10);
            if (!this.al_ClientName.Contains(ClientName))
            {
                this.al_ClientName.Add(ClientName);
            }
            ClientState cs = new ClientState();
            cs.ClientName = ClientName;
            cs.ClientSocket = sock;
            cs.RWL_ClientSock = new ReaderWriterLock();
            cs.bsBuffer = new byte[1024];
            this.RWL_sl_ClientState_ClientName.AcquireWriterLock(-1);
            if (!this.sl_ClientState_ClientName.ContainsKey(ClientName))
            {
                this.sl_ClientState_ClientName.Add(ClientName, cs);
            }
            else
            {
                //系统曾经有连接
                ClientState _Client = (ClientState)this.sl_ClientState_ClientName[ClientName];
                this.sl_ClientState_ClientName.Remove(ClientName);
                System.Threading.Interlocked.Increment(ref _Client.StateFlag);//提示其他线程不需要再清理资源
                try
                {
                    _Client.ClientSocket.Close();
                }
                catch
                { }
                _Client.RWL_ClientSock.ReleaseLock();
                EventArgsItem eia = new EventArgsItem();
                eia.Type = EventArgsItemType.ClientLogOutEventArgs;
                ClientLogOutEventArgs eea = new ClientLogOutEventArgs();
                eea.LogoutTime = DateTime.Now;
                eea.ClientName = ClientName;
                eia.Item = eea;
                this.al_ClientEventArgs.Add(eia);
                this.sl_ClientState_ClientName.Add(ClientName, cs);
            }
            this.RWL_sl_ClientState_ClientName.ReleaseLock();
            //声称登陆事件
            EventArgsItem eai = new EventArgsItem();
            eai.Type = EventArgsItemType.ClientLoginEventArgs;
            ClientLoginEventArgs ea = new ClientLoginEventArgs();
            ea.ClientIPEP = ClientIPEP;
            ea.Message = new byte[size];
            Array.Copy(bsmsg, 0, ea.Message, 0, size);
            ea.LoginDate = DateTime.Now;
            ea.Name = ClientName;
            eai.Item = ea;
            this.al_ClientEventArgs.Add(eai);
            //下面开始接收数据
            this.WriteLog("准备开始异步接收从" + ClientName + "发送来的数据IP:" + ClientIPEP.Address.ToString() + ":" + ClientIPEP.Port.ToString());
            cs.RWL_ClientSock.AcquireWriterLock(-1);
            try
            {
                cs.ClientSocket.BeginReceive(cs.bsBuffer, 0, 1024, SocketFlags.None, new AsyncCallback(AsyncReceiveClientTCPMessge), cs);
            }
            catch
            { }
            cs.RWL_ClientSock.ReleaseLock();
            this.ARE_Listen.Set();        }

Answer 3

        private void ReceiveClientUDPMessage()
        {
            IPEndPoint ServerIPEP = new IPEndPoint(this.ServerIP, this.Port);
            while (this.IFStart)
            {
                try
                {
                    EventArgsItem eai = new EventArgsItem();
                    eai.Type = EventArgsItemType.ReceiveClientUDPMessageEventArgs;
                    ReceiveClientUDPMessageEventArgs ea = new ReceiveClientUDPMessageEventArgs();
                    ea.Message = new byte[bsmsg.Length];
                    Array.Copy(bsmsg, 0, ea.Message, 0, bsmsg.Length);
                    ea.ReceiveTime = DateTime.Now;
                    ea.Name = bsmsg[1].ToString("X") + bsmsg[2].ToString("X") + bsmsg[3].ToString("X") + bsmsg[4].ToString("X") + bsmsg[5].ToString("X");
                    eai.Item = ea;
                    string strtemp = "";
                    for (int i = 0; i < bsmsg.Length; i++)
                    {
                        strtemp += bsmsg[i].ToString("X2");
                    }
                    this.WriteTemp(strtemp);
                }
                catch (Exception ex)
                {
                    string str = ex.Message;
                    this.WriteLog("接收UDP数据异常,原因-->" + ex.Message);
                }
            }
        }
        

Answer 4

自己写load test工具啊，怎么不搞一个test tool啊，你要求的这样功能一般test tool都有。

Answer 5

[code=C#]
        private void ReceiveClientUDPMessage()
        {
            IPEndPoint ServerIPEP = new IPEndPoint(this.ServerIP, this.Port);
            while (this.IFStart)
            {
                try
                {
                    EventArgsItem eai = new EventArgsItem();
                    eai.Type = EventArgsItemType.ReceiveClientUDPMessageEventArgs;
                    ReceiveClientUDPMessageEventArgs ea = new ReceiveClientUDPMessageEventArgs();
                    ea.Message = new byte[bsmsg.Length];
                    Array.Copy(bsmsg, 0, ea.Message, 0, bsmsg.Length);
                    ea.ReceiveTime = DateTime.Now;
                    ea.Name = bsmsg[1].ToString("X") + bsmsg[2].ToString("X") + bsmsg[3].ToString("X") + bsmsg[4].ToString("X") + bsmsg[5].ToString("X");
                    eai.Item = ea;
                    string strtemp = "";
                    for (int i = 0; i < bsmsg.Length; i++)
                    {
                        strtemp += bsmsg[i].ToString("X2");
                    }
                    this.WriteTemp(strtemp);
                }
                catch (Exception ex)
                {
                    string str = ex.Message;
                    this.WriteLog("接收UDP数据异常,原因-->" + ex.Message);
                }
            }
        }
        
[/code]

Answer 6

代码贴的有点乱
你主要看
ThreadListen()方法开始侦听连接
AcceptCallBack()异步接收数据并为每个连接创建Socket
AsyncReceiveClientTCPMessge()每个连接的异步接收数据不明白的地方直接问我另外还有个封装连接的类
public class ClientState
    {
        public string ClientName;
        public Socket ClientSocket;
        public ReaderWriterLock RWL_ClientSock;
        public byte[] bsBuffer;
        public int StateFlag = 1;
        public DateTime LastReceiveClientMessageTime;
        private ReaderWriterLock RWL_LastReceiveClientMessageTime;
        public int GetNoDataTimeSpan()
        {
            this.RWL_LastReceiveClientMessageTime.AcquireReaderLock(-1);
            TimeSpan ts = DateTime.Now - this.LastReceiveClientMessageTime;
            this.RWL_LastReceiveClientMessageTime.ReleaseReaderLock();
            return (int)ts.TotalSeconds;
        }
        public void SetLastRecvClientMessageTime()
        {
            this.RWL_LastReceiveClientMessageTime.AcquireWriterLock(-1);
            this.LastReceiveClientMessageTime = DateTime.Now;
            this.RWL_LastReceiveClientMessageTime.ReleaseLock();
        }
        //public byte[] LoginMessage;
        public ClientState()
        {
            this.bsBuffer = new byte[1024];
            this.LastReceiveClientMessageTime = DateTime.Now;
            this.RWL_LastReceiveClientMessageTime = new ReaderWriterLock();
        }
    }