- A+
所属分类：.NET技术
服务端
工作需要又需要用到socketTCP通讯，这么多年了，终于稍微能写点了。让我说其实也说不出个啥来，看了很多的异步后稍微对异步socket的导流 endreceive后 再beginreceive 形成一个内循环有了个认识，加上我自己的封包拆包机制，然后再仿那些其它的大多数代码结构弄点onReceive事件进行 收包触发。整个过程就算差不多了 ，基本上是能够可靠运行的 靠谱的 中规中矩的，要说啥创新读到嘛真的谈不上。代码中写了很多low逼注释 也是为了方便自己理解 请无视。下面是server端代码，使用异步机制accept 异步receive ，成员有 clients代表当前在线的客户端 客户端socket包装为EndpointClient ，有onClientAddDel 代表客户端上线掉线事件，有onReceive代表所有客户端的收包事件，clients由于是异步的多线程访问就要涉及多线程管控 所以使用lock ，服务端有sendToAll() 和SendToSomeOne（）毫无疑问这也是通过调用特定的clients来做的。
以下是服务端代码
1 public class MsgServerSchedule 2 { 3 4 5 Socket serverSocket; 6 public Action<List<string>> onClientAddDel; 7 public Action<Telegram_Base> onReceive; 8 bool _isRunning = false; 9 10 11 int port; 12 13 public TelgramType telType; 14 15 static List<EndpointClient> clients; 16 17 public bool isRunning { get { return _isRunning; } } 18 public MsgServerSchedule(int _port) 19 { 20 //any 就决定了 ip地址格式是v4 21 //IPEndPoint endPoint = new IPEndPoint(IPAddress.Any, 7654); 22 //socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); 23 24 this.port = _port; 25 26 clients = new List<EndpointClient>(); 27 28 Console.WriteLine("constructor"); 29 30 } 31 32 public void Start() 33 { 34 try 35 { 36 IPEndPoint endPoint = new IPEndPoint(IPAddress.Any, port); 37 serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); 38 serverSocket.Bind(endPoint); 39 serverSocket.Listen(port); 40 41 serverSocket.BeginAccept(new AsyncCallback(AcceptCallback), serverSocket); 42 43 _isRunning = true; 44 Console.WriteLine("start"); 45 } 46 catch (Exception ex) 47 { 48 _isRunning = false; 49 serverSocket = null; 50 51 Console.WriteLine("服务启动出现错误,可能端口已被占用："+port); 52 Console.WriteLine(ex.Message); 53 } 54 55 } 56 57 public void Stop() 58 { 59 for (int i = 0; i < clients.Count; i++) 60 { 61 clients[i].Close(); 62 } 63 ClientAddDelGetList(null, EndPointClientsChangeType.ClearAll); 64 serverSocket.Close(); 65 _isRunning = false; 66 } 67 68 public void SendToAll(Telegram_Base tel) 69 { 70 for (int i = 0; i < clients.Count; i++) 71 { 72 clients[i].Send(tel); 73 } 74 } 75 76 public void SendToSomeOne(Telegram_Base tel) 77 { 78 for (int i = 0; i < clients.Count; i++) 79 { 80 if(clients[i].remoteIPPort==tel.remoteIPPort) 81 { 82 clients[i].Send(tel); 83 break; 84 } 85 } 86 } 87 88 //新增与删除客户端 秉持原子操作 89 List<string> ClientAddDelGetList(EndpointClient cli, EndPointClientsChangeType changeType) 90 { 91 //异步同时有新客户端上线 与下线 不进行资源互斥访问会报错 92 lock (this) 93 { 94 if (changeType == EndPointClientsChangeType.Add) 95 { 96 clients.Add(cli); 97 } 98 else if(changeType== EndPointClientsChangeType.Del) 99 { 100 var beRemoveClient = clients.First(r => r.remoteIPPort == cli.remoteIPPort); 101 if (beRemoveClient != null) 102 clients.Remove(beRemoveClient); 103 } 104 else if(changeType== EndPointClientsChangeType.ClearAll) 105 { 106 clients.Clear(); 107 } 108 else if (changeType == EndPointClientsChangeType.GetAll) 109 { 110 List<string> onLines = new List<string>(clients.Count); 111 for (int i = 0; i < clients.Count; i++) 112 { 113 onLines.Add(clients[i].remoteIPPort); 114 } 115 return onLines; 116 } 117 else 118 { 119 return null; 120 } 121 } 122 return null; 123 } 124 //异步监听客户端 有客户端到来时的回调 125 private void AcceptCallback(IAsyncResult iar) 126 { 127 //server端一直在receive 能够感知到客户端掉线 （连上后 关闭客户端 server立即有错误爆出） 128 //但是同情况 关闭server端 客户端无错误爆出 直到点发送 才有错误爆出 129 //由此得出 处于receive才会有掉线感知 ，send时发现发不出去自然也会有感知 跟人的正常思维理解是一样的 130 //虽然tcp是所谓的长连接 ，通过反复测试 ->但是双方相互都处在一个静止状态 是无法 确定在不在的 131 //连上后平常的情况下 并没有数据流通 的 ,双方只是一个状态的保持而已。 132 //这也是为什么 好多服务 客户端 程序 都有个心跳机制（由此我们可以想到继续完善 弄个客户端列表 心跳超时的剔除列表 正常发消息send 或receive 异常的剔除列表 删除clientSocket 133 //其实非要说吧 一般情况 服务端一直在receive 不用心跳其实也是可以的（客户端可能是真的需要 134 //tcp底层就已经有了一个判断对方在不在的机制 ， 对方直接关程序 结束进程 这些 只要tcp在receive就立即能够感知 所以说心跳 用不用看情况吧 135 136 //tcp 不会丢包 哪怕是连接建立了 你还没开始receive 对方却先发了， 137 //对方只要是发了的数据 都由操作系统像个缓存样给你放那的 不会掉 你再隔10秒开始receive都能rec的到 138 139 //tcp甚至在拔掉网线 再重新插上 都可以保证数据一致性 140 //tcp的包顺序能够保证 先发的先到 141 142 //nures代码中那些beginreceivexxx 异步receive的核心机制就是 ，假定数据到的时候把数据保存到xxx数组 143 //真正endreceive的时候 其实数据已经接收 处理完成了 144 145 try 146 { 147 148 //处理完当前accept 149 Socket currentSocket = serverSocket.EndAccept(iar); 150 151 EndpointClient client = new EndpointClient(currentSocket,telType); 152 153 //新增客户端 154 ClientAddDelGetList(client, EndPointClientsChangeType.Add); 155 156 if (onClientAddDel != null) 157 { 158 List<string> onlines = ClientAddDelGetList(null, EndPointClientsChangeType.GetAll); 159 onClientAddDel(onlines); 160 161 //客户端异常掉线 162 client.onClientDel = new Action<string>((_remoteIPPort) => 163 { 164 ClientAddDelGetList(new EndpointClient(){ remoteIPPort=_remoteIPPort} , EndPointClientsChangeType.Del); 165 166 List<string> onlines2 = ClientAddDelGetList(null, EndPointClientsChangeType.GetAll); 167 onClientAddDel(onlines2); 168 }); 169 } 170 171 172 173 //这句到时调用完成后 就会自动把 receivebuffer填充 //要接收的字节数 系统底层操作一次接收多少字节 其实意义不大 174 //总是从0开始（就是说并发时会覆盖 175 176 Console.WriteLine(string.Format("new client ->{0}", currentSocket.RemoteEndPoint.ToString())); 177 178 //currentSocket.Close(); 179 //Application.Exit(); 180 181 //Thread.Sleep(1000 * 10); 182 client.onReceive += this.onReceive; 183 184 client.BeginReceive(); 185 186 187 //立即开始accept新的客户端 188 if (isRunning == true && serverSocket != null) 189 serverSocket.BeginAccept(AcceptCallback, serverSocket); 190 //beginAccept 最开始的方法可以不一样 ，但最终肯定是一个不断accept的闭环过程 191 //整个过程就像个导流样 ，最开始用异步导流到一个固定的点 然后让其循环源源不断运转 192 193 //加asynccallback 有什么不一样么 194 //socket.BeginAccept(new AsyncCallback( AcceptCallback), socket); 195 196 } 197 catch (Exception ex) 198 { 199 Console.WriteLine("AcceptCallback Error"); 200 Console.WriteLine(ex.Message); 201 } 202 203 } 204 205 206 }
EndpointClient终端代码代表客户端的对口人，他的onReceive 等资源从服务端继承 ，如果服务端想给某个特定客户端发数据则会调用他们中的某一个 毫无疑问这是通过remoteIPport来判断的，这些都是编写基本socket结构轻车熟路的老套路
以下EndpointClient代码
1 public class EndpointClient 2 { 3 Socket workingSocket; 4 static int receiveBufferLenMax = 5000; 5 byte[] onceReadDatas = new byte[receiveBufferLenMax]; 6 List<byte> receiveBuffer = new List< byte>(receiveBufferLenMax); 7 8 public string remoteIPPort { get; set; } 9 10 //当前残留数据区 接收数据的起始指针（也代表缓冲区数据长度 11 int receiveBufferLen = 0; 12 13 14 TelgramType telType; 15 16 public Action<Telegram_Base> onReceive; 17 public Action<string> onClientDel; 18 19 public EndpointClient() 20 { 21 22 } 23 public EndpointClient(Socket _socket,TelgramType _telType) 24 { 25 this.remoteIPPort = _socket.RemoteEndPoint.ToString(); 26 this.telType = _telType; 27 workingSocket = _socket; 28 } 29 30 public void Send(Telegram_Base tel) 31 { 32 //try 33 //{ 34 if(workingSocket==null) 35 { 36 Console.WriteLine("未初始化的EndpointClient"); 37 return; 38 } 39 if (tel is Telegram_Schedule) 40 { 41 Telegram_Schedule telBeSend = tel as Telegram_Schedule; 42 if (telBeSend.dataBytes.Length != telBeSend.dataLen) 43 { 44 Console.WriteLine("尝试发送数据长度格式错误的报文"); 45 return; 46 } 47 48 byte[] sendBytesHeader = telBeSend.dataBytesHeader; 49 byte[] sendbytes = telBeSend.dataBytes; 50 51 //数据超过缓冲区长度 会导致无法拆包 52 if (sendbytes.Length <= receiveBufferLenMax) 53 { 54 workingSocket.BeginSend(sendBytesHeader, 0, sendBytesHeader.Length, 0, null, null); 55 workingSocket.BeginSend(sendbytes, 0, sendbytes.Length, 0,null,null 56 57 ); 58 } 59 else 60 { 61 Console.WriteLine("发送到调度客户端的数据超过缓冲区长度"); 62 throw new Exception("发送到调度客户端的数据超过缓冲区长度"); 63 } 64 65 } 66 else if (tel is Telegram_SDBMsg) 67 { 68 69 } 70 71 //} 72 //catch (Exception ex) 73 //{ 74 75 // Console.WriteLine(ex.Message); 76 // throw ex; 77 //} 78 } 79 80 public void BeginReceive() 81 { 82 if (workingSocket == null) 83 { 84 Console.WriteLine("未初始化的EndpointClient"); 85 return; 86 } 87 88 receiveBufferLen = 0; 89 workingSocket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax, SocketFlags.None, 90 ReceiveCallback, 91 this); 92 } 93 private void ReceiveCallback(IAsyncResult iar) 94 { 95 try 96 { 97 EndpointClient cli = (EndpointClient)iar.AsyncState; 98 Socket socket = cli.workingSocket; 99 int reads = socket.EndReceive(iar); 100 101 if (reads > 0) 102 { 103 104 for (int i = 0; i < reads; i++) 105 { 106 receiveBuffer.Add(onceReadDatas[i]); 107 } 108 109 //具体填充了多少看返回值 此时 数据已经在buffer中了 110 receiveBufferLen += reads; 111 //加完了后解析 阻塞式处理 结束后开启新的接收 112 SloveTelData(); 113 114 if (receiveBufferLenMax - receiveBufferLen > 0) 115 { 116 //接收完了 继续beginreceive 开启异步的下次接收 (如果缓冲区有残留数据 则接收长度变短 ，没接收到的让其留在socket不会丢失 下次接收) 117 socket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax - receiveBufferLen, SocketFlags.None, ReceiveCallback, this); 118 } 119 else//阻塞式处理都完成一遍了 都还没清理出任何缓冲区空间 毫无疑问 整体运转机制已经挂了 不用beginreceive下一次了 120 { 121 Close(); 122 //移除自己 123 if (onClientDel != null) 124 { 125 onClientDel(remoteIPPort); 126 } 127 Console.WriteLine("服务端接口解析数据出现异常"); 128 throw new Exception("服务端接口解析数据出现异常"); 129 } 130 } 131 else//reads==0 客户端已关闭 132 { 133 Close(); 134 //移除自己 135 if (onClientDel != null) 136 { 137 onClientDel(remoteIPPort); 138 } 139 } 140 } 141 catch (Exception ex) 142 { 143 Close(); 144 //移除自己 145 if (onClientDel != null) 146 { 147 onClientDel(remoteIPPort); 148 } 149 150 Console.WriteLine("ReceiveCallback Error"); 151 Console.WriteLine(ex.Message); 152 } 153 154 } 155 void SloveTelData() 156 { 157 //进行数据解析 158 SloveTelDataUtil slo = new SloveTelDataUtil(); 159 160 if (telType == TelgramType.Schedule) 161 { 162 List<Telegram_Schedule> dataEntitys = slo.Slove_Telegram_Schedule(receiveBuffer, receiveBufferLen, this.remoteIPPort); 163 //buffer已经被处理一遍了 使用新的长度 164 receiveBufferLen = receiveBuffer.Count; 165 //解析出的每一个对象都触发 onreceive 166 for (int i = 0; i < dataEntitys.Count; i++) 167 { 168 if (onReceive != null) 169 onReceive(dataEntitys[i]); 170 } 171 } 172 else if (telType == TelgramType.SDBMsg) 173 { 174 175 } 176 177 } 178 179 180 public void Close() 181 { 182 try 183 { 184 receiveBuffer.Clear(); 185 receiveBufferLen = 0; 186 if (workingSocket != null && workingSocket.Connected) 187 workingSocket.Close(); 188 } 189 catch (Exception ex) 190 { 191 Console.WriteLine(ex.Message); 192 } 193 194 } 195 }
数据拆包与封包粘包处理
上面的代码可以看到 数据包处理都在receiveCallback里 SloveTelData，也是通用的套路 ，解析到完整的包后从缓冲区移除 解析多少个包触发多少次事件，残余数据留在缓冲区 然后继续开始新的beginReceive往缓冲区加。在异步机制中 到达endReceive的时候数据已经在缓冲区里了，这个自不用多说噻。数据包和粘包逻辑在公共类库里供客户端服务端共同调用
以下是粘包处理逻辑
1 public class SloveTelDataUtil 2 { 3 List<Telegram_Schedule> solveList; 4 public SloveTelDataUtil() 5 { 6 } 7 8 List<byte> buffer; 9 int bufferLen; 10 int bufferIndex = 0; 11 string remoteIPPort; 12 public List<Telegram_Schedule> Slove_Telegram_Schedule( List< byte> _buffer,int _bufferLen,string _remoteIPPort) 13 { 14 15 solveList = new List<Telegram_Schedule>(); 16 17 bufferIndex = 0; 18 19 buffer = _buffer; 20 bufferLen = _bufferLen; 21 remoteIPPort = _remoteIPPort; 22 23 //小于最小长度 直接返回 24 if (bufferLen < 12) 25 return solveList; 26 27 //进行数据解析 28 bool anaysisOK = false; 29 while (anaysisOK=AnaysisData_Schedule()==true)//直到解析的不能解析为止 30 { 31 } 32 return solveList; 33 } 34 35 public bool AnaysisData_Schedule() 36 { 37 if (bufferLen - bufferIndex < GlobalSymbol.Headerlen) 38 return false; 39 40 //解析出一个数据对象 41 Telegram_Schedule telObj = new Telegram_Schedule(); 42 43 //必定是大于最小数据大小的 44 telObj.dataBytesHeader = new byte[GlobalSymbol.Headerlen]; 45 buffer.CopyTo(bufferIndex, telObj.dataBytesHeader, 0, GlobalSymbol.Headerlen); 46 47 byte[] btsHeader = new byte[4]; 48 byte[] btsCommand = new byte[4]; 49 byte[] btsLen = new byte[4]; 50 51 btsHeader[0] = buffer[bufferIndex]; 52 btsHeader[1] = buffer[bufferIndex+1]; 53 btsHeader[2] = buffer[bufferIndex+2]; 54 btsHeader[3] = buffer[bufferIndex+3]; 55 56 bufferIndex += 4; 57 58 btsCommand[0] = buffer[bufferIndex]; 59 btsCommand[1] = buffer[bufferIndex + 1]; 60 btsCommand[2] = buffer[bufferIndex + 2]; 61 btsCommand[3] = buffer[bufferIndex + 3]; 62 63 bufferIndex += 4; 64 65 btsLen[0] = buffer[bufferIndex]; 66 btsLen[1] = buffer[bufferIndex + 1]; 67 btsLen[2] = buffer[bufferIndex + 2]; 68 btsLen[3] = buffer[bufferIndex + 3]; 69 70 bufferIndex += 4; 71 72 73 74 int dataLen = BitConverter.ToInt32(btsLen, 0); 75 telObj.header = BitConverter.ToUInt32(btsHeader, 0); 76 telObj.command = BitConverter.ToInt32(btsCommand, 0); 77 telObj.remoteIPPort = remoteIPPort; 78 79 if(dataLen>0) 80 { 81 //数据区小于得到的数据长度 说明数据部分还没接收到 不删除缓冲区 不做任何处理 82 //下次来了连着头一起解析 83 if (bufferLen - GlobalSymbol.Headerlen < dataLen) 84 { 85 86 bufferIndex -= 12;// 87 88 89 return false; 90 91 } 92 else 93 { 94 95 telObj.dataLen = dataLen; 96 telObj.dataBytes = new byte[dataLen]; 97 buffer.CopyTo(bufferIndex, telObj.dataBytes, 0, dataLen); 98 99 solveList.Add(telObj); 100 //bufferIndex += dataLen; 101 102 //解析成功一次 移除已解析的 103 for (int i = 0; i < GlobalSymbol.Headerlen+dataLen; i++) 104 { 105 buffer.RemoveAt(0); 106 } 107 bufferIndex = 0; 108 bufferLen = buffer.Count; 109 return true; 110 } 111 } 112 else 113 { 114 115 telObj.dataLen = 0; 116 solveList.Add(telObj); 117 //bufferIndex += 0; 118 //解析成功一次 移除已解析的 119 for (int i = 0; i < GlobalSymbol.Headerlen; i++) 120 { 121 buffer.RemoveAt(0); 122 } 123 //解析成功一次因为移除了缓冲区 bufferIndex置0 124 bufferIndex = 0; 125 bufferLen = buffer.Count; 126 return true; 127 } 128 129 } 130 131 132 public List<Telegram_SDBMsg> Slove_Telegram_SDBMsg(ref byte[] buffer) 133 { 134 return new List<Telegram_SDBMsg>(); 135 } 136 }
我们看到用到的数据包对象是Telegram_Schedule ，中间保存有报文数据，数据发送的目标等信息。
以下是数据包结构代码
1 public class Telegram_Base 2 { 3 public string remoteIPPort { get; set; } 4 //数据内容 5 public byte[] dataBytes { get; set; } 6 //头部内容的序列化 7 public byte[] dataBytesHeader { get; set; } 8 9 public string jsonStr { get; set; } 10 virtual public void SerialToBytes() 11 { 12 13 } 14 15 virtual public void SloveToTel() 16 { 17 18 } 19 20 } 21 22 public class Telegram_Schedule:Telegram_Base 23 { 24 25 //头部标识 4字节 26 public UInt32 header { get; set; } 27 //命令对应枚举的 int 4字节 28 public int command { get; set; } 29 //数据长度 4字节 30 public int dataLen { get; set; } 31 32 33 34 override public void SerialToBytes() 35 { 36 //有字符串数据 但是待发送字节是空 37 if ((string.IsNullOrEmpty(jsonStr) == false ))//&& (dataBytes==null || dataBytes.Length==0) 38 { 39 dataBytes = Encoding.UTF8.GetBytes(jsonStr); 40 dataLen = dataBytes.Length; 41 dataBytesHeader = new byte[GlobalSymbol.Headerlen]; 42 43 header = GlobalSymbol.HeaderSymbol; 44 45 byte[] btsHeader = BitConverter.GetBytes(header); 46 byte[] btsCommand = BitConverter.GetBytes(command); 47 byte[] btsLen = BitConverter.GetBytes(dataLen); 48 49 Array.Copy(btsHeader, 0, dataBytesHeader, 0, 4); 50 Array.Copy(btsCommand, 0, dataBytesHeader, 4, 4); 51 Array.Copy(btsLen, 0, dataBytesHeader, 8, 4); 52 53 } 54 else if((string.IsNullOrEmpty(jsonStr) == true )&& (dataBytes==null || dataBytes.Length==0)){ 55 dataLen = 0; 56 dataBytes = new byte[0]; 57 58 dataBytesHeader = new byte[GlobalSymbol.Headerlen]; 59 60 header = GlobalSymbol.HeaderSymbol; 61 62 byte[] btsHeader = BitConverter.GetBytes(header); 63 byte[] btsCommand = BitConverter.GetBytes(command); 64 byte[] btsLen = BitConverter.GetBytes(dataLen); 65 66 Array.Copy(btsHeader, 0, dataBytesHeader, 0, 4); 67 Array.Copy(btsCommand, 0, dataBytesHeader, 4, 4); 68 Array.Copy(btsLen, 0, dataBytesHeader, 8, 4); 69 } 70 } 71 72 override public void SloveToTel() 73 { 74 //只解析字符串数据部分 ，header 和len 在接收之初就已解析 75 this.jsonStr = Encoding.UTF8.GetString(this.dataBytes); 76 } 77 78 }
客户端代码
最后是客户端，有了上面的结构，客户端就不足为谈了，稍微了解socket的人都熟知套路的 基本跟EndpointClient一致
1 public class MsgClientSchedule 2 { 3 Socket workingSocket; 4 //缓冲区最大数据长度 5 static int receiveBufferLenMax = 5000; 6 //单次receive数据（取决于tcp底层封包 但是不会超过缓冲区最大长度 7 byte[] onceReadDatas = new byte[receiveBufferLenMax]; 8 //未解析到完整数据包时的残余数据保存区 9 List<byte> receiveBuffer = new List<byte>(receiveBufferLenMax); 10 11 string serverIP { get; set; } 12 int serverPort { get; set; } 13 public string localIPPort { get; set; } 14 15 //残余缓冲区数据长度 16 int receiveBufferLen = 0; 17 18 bool _isConnected { get; set; } 19 20 TelgramType telType; 21 22 //收一个包时触发 23 public Action<Telegram_Base> onReceive; 24 //与服务端断链时触发 25 public Action<string> onClientDel; 26 27 28 public bool isConnected { get { return _isConnected; } } 29 public MsgClientSchedule(string _serverIP,int _port) 30 { 31 serverIP = _serverIP; 32 serverPort = _port; 33 _isConnected = false; 34 } 35 36 public void Connect() 37 { 38 try 39 { 40 workingSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.IP); 41 IPEndPoint ipport = new IPEndPoint(IPAddress.Parse(serverIP), serverPort); 42 workingSocket.Connect(ipport); 43 44 localIPPort = workingSocket.LocalEndPoint.ToString(); 45 _isConnected = true; 46 BeginReceive(); 47 } 48 catch (Exception ex) 49 { 50 workingSocket = null; 51 _isConnected = false; 52 53 Console.WriteLine(ex.Message); 54 } 55 56 } 57 58 59 60 61 public void Send(Telegram_Base tel) 62 { 63 try 64 { 65 if(_isConnected==false) 66 { 67 Console.WriteLine("未连接到服务器"); 68 return; 69 } 70 if (tel is Telegram_Schedule) 71 { 72 Telegram_Schedule telBeSend = tel as Telegram_Schedule; 73 if (telBeSend.dataBytes.Length != telBeSend.dataLen) 74 { 75 Console.WriteLine("尝试发送数据长度格式错误的报文"); 76 return; 77 } 78 byte[] sendBytesHeader = telBeSend.dataBytesHeader; 79 byte[] sendbytes = telBeSend.dataBytes; 80 81 //数据超过缓冲区长度 会导致无法拆包 82 if (sendbytes.Length <= receiveBufferLenMax) 83 { 84 workingSocket.BeginSend(sendBytesHeader, 0, sendBytesHeader.Length, 0, null, null); 85 workingSocket.BeginSend(sendbytes, 0, sendbytes.Length, 0, null, null 86 87 ); 88 } 89 else 90 { 91 Console.WriteLine("发送到调度客户端的数据超过缓冲区长度"); 92 throw new Exception("发送到调度客户端的数据超过缓冲区长度"); 93 } 94 95 96 } 97 else if (tel is Telegram_SDBMsg) 98 { 99 100 } 101 102 } 103 catch (Exception ex) 104 { 105 Close(); 106 Console.WriteLine(ex.Message); 107 //throw ex; 108 } 109 } 110 111 public void BeginReceive() 112 { 113 receiveBufferLen = 0; 114 workingSocket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax, SocketFlags.None, 115 ReceiveCallback, 116 117 this); 118 } 119 private void ReceiveCallback(IAsyncResult iar) 120 { 121 try 122 { 123 MsgClientSchedule cli = (MsgClientSchedule)iar.AsyncState; 124 Socket socket = cli.workingSocket; 125 int reads = socket.EndReceive(iar); 126 127 if (reads > 0) 128 { 129 130 for (int i = 0; i < reads; i++) 131 { 132 receiveBuffer.Add(onceReadDatas[i]); 133 } 134 135 //具体填充了多少看返回值 此时 数据已经在buffer中了 136 137 receiveBufferLen += reads; 138 139 //加完了后解析 阻塞式处理 结束后开启新的接收 140 SloveTelData(); 141 142 143 144 if (receiveBufferLenMax - receiveBufferLen > 0) 145 { 146 //接收完了 继续beginreceive 开启异步的下次接收 (如果缓冲区有残留数据 则接收长度变短 ，没接收到的让其留在socket不会丢失 下次接收) 147 socket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax - receiveBufferLen, SocketFlags.None, ReceiveCallback, this); 148 } 149 else//阻塞式处理都完成一遍了 都还没清理出任何缓冲区空间 毫无疑问 整体运转机制已经挂了 不用beginreceive下一次了 150 { 151 Close(); 152 153 Console.WriteLine("服务端接口解析数据出现异常"); 154 throw new Exception("服务端接口解析数据出现异常"); 155 } 156 } 157 else//reads==0客户端已关闭 158 { 159 Close(); 160 } 161 } 162 catch (Exception ex) 163 { 164 Close(); 165 166 Console.WriteLine("ReceiveCallback Error"); 167 Console.WriteLine(ex.Message); 168 } 169 170 } 171 private void SloveTelData() 172 { 173 174 //进行数据解析 175 SloveTelDataUtil slo = new SloveTelDataUtil(); 176 177 if (telType == TelgramType.Schedule) 178 { 179 List<Telegram_Schedule> dataEntitys = slo.Slove_Telegram_Schedule(receiveBuffer, receiveBufferLen,serverIP+":"+serverPort.ToString()); 180 //buffer已经被处理一遍了 使用新的长度 181 receiveBufferLen = receiveBuffer.Count; 182 //解析出的每一个对象都触发 onreceive 183 for (int i = 0; i < dataEntitys.Count; i++) 184 { 185 if (onReceive != null) 186 onReceive(dataEntitys[i]); 187 } 188 } 189 else if (telType == TelgramType.SDBMsg) 190 { 191 192 } 193 194 } 195 196 197 public void Close() 198 { 199 try 200 { 201 _isConnected = false; 202 203 receiveBuffer.Clear(); 204 receiveBufferLen = 0; 205 if (workingSocket != null && workingSocket.Connected) 206 workingSocket.Close(); 207 } 208 catch (Exception ex) 209 { 210 Console.WriteLine(ex.Message); 211 } 212 213 } 214 215 }
服务端调用
构建一个winform基本项目
1 List<string> clients; 2 MsgServerSchedule server; 3 private void btn_start_Click(object sender, EventArgs e) 4 { 5 server = new MsgServerSchedule(int.Parse(tbx_port.Text)); 6 7 8 server.Start(); 9 if (server.isRunning == true) 10 { 11 btn_start.Enabled = false; 12 13 server.onReceive += new Action<Telegram_Base>( 14 (tel) => 15 { 16 this.BeginInvoke(new Action(() => 17 { 18 if (tel is Telegram_Schedule) 19 { 20 Telegram_Schedule ts = tel as Telegram_Schedule; 21 ts.SloveToTel(); 22 Console.WriteLine(string.Format("commandType:{0}", ((ScheduleTelCommandType)ts.command).ToString())); 23 24 tbx_msgs.Text += ts.remoteIPPort + ">" + ts.jsonStr + "rn"; 25 26 //数据回发测试 27 string fromip = ts.remoteIPPort; 28 string srcMsg = ts.jsonStr; 29 string fromServerMsg = ts.jsonStr + " -from server"; 30 ts.jsonStr = fromServerMsg; 31 32 33 //如果消息里有指向信息 则转送到对应的客户端 34 if (clients != null) 35 { 36 string to = null; 37 for (int i = 0; i < clients.Count; i++) 38 { 39 if (srcMsg.Contains(clients[i])) 40 { 41 to = clients[i]; 42 break; 43 } 44 } 45 46 if (to != null) 47 { 48 ts.remoteIPPort = to; 49 string toMsg; 50 //toMsg= srcMsg.Replace(to, ""); 51 toMsg = srcMsg.Replace(to, fromip); 52 ts.jsonStr = toMsg; 53 ts.SerialToBytes(); 54 55 server.SendToSomeOne(ts); 56 } 57 else 58 { 59 ts.SerialToBytes(); 60 server.SendToSomeOne(ts); 61 } 62 } 63 } 64 })); 65 66 } 67 ); 68 69 server.onClientAddDel += new Action<List<string>>((onlines) => 70 { 71 this.BeginInvoke( 72 new Action(() => 73 { 74 clients = onlines; 75 listbox_clients.Items.Clear(); 76 77 for (int i = 0; i < onlines.Count; i++) 78 { 79 listbox_clients.Items.Add(onlines[i]); 80 } 81 })); 82 }); 83 } 84 } 85 private void btn_sendAll_Click(object sender, EventArgs e) 86 { 87 Telegram_Schedule tel = new Telegram_Schedule(); 88 tel.header = GlobalSymbol.HeaderSymbol; 89 tel.command = (int)ScheduleTelCommandType.StartC2S; 90 tel.jsonStr = tbx_sendAll.Text; 91 tel.SerialToBytes(); 92 93 server.SendToAll(tel); 94 }
客户端调用
1 MsgClientSchedule client; 2 3 private void btn_start_Click(object sender, EventArgs e) 4 { 5 client = new MsgClientSchedule(tbx_ip.Text, int.Parse(tbx_port.Text)); 6 7 client.Connect(); 8 9 if (client.isConnected == true) 10 { 11 btn_start.Enabled = false; 12 13 label1.Text = client.localIPPort; 14 15 client.onReceive = new Action<Telegram_Base>((tel) => 16 { 17 this.BeginInvoke( 18 new Action(() => 19 { 20 tel.SloveToTel(); 21 tbx_rec.Text += tel.jsonStr + "rn"; 22 23 })); 24 }); 25 } 26 27 } 28 29 30 31 private void btn_send_Click(object sender, EventArgs e) 32 { 33 34 if (client == null || client.isConnected == false) 35 return; 36 37 //for (int i = 0; i < 2; i++) 38 //{ 39 Telegram_Schedule tel = new Telegram_Schedule(); 40 tel.command = (int)ScheduleTelCommandType.MsgC2S; 41 42 tel.jsonStr = tbx_remoteip.Text+">"+ tbx_msgSend.Text; 43 tel.SerialToBytes();//发出前要先序列化 44 45 client.Send(tel); 46 //} 47 48 }
实现效果
可以多客户端连接互相自由发送消息，服务端可以编写转发规则代码，那些什么棋牌啊 互动白板 以及其他类似的应用就可以基于此之上发挥想象了
