服务端

基本的结构

工作需要又需要用到socketTCP通讯，这么多年了，终于稍微能写点了。让我说其实也说不出个啥来，看了很多的异步后稍微对异步socket的导流 endreceive后再beginreceive 形成一个内循环有了个认识，加上我自己的封包拆包机制，然后再仿那些其它的大多数代码结构弄点onReceive事件进行收包触发。整个过程就算差不多了，基本上是能够可靠运行的靠谱的中规中矩的，要说啥创新独到嘛真的谈不上。代码中写了很多low逼注释也是为了方便自己理解请无视。下面是server端代码，使用异步机制accept 异步receive ，成员有 clients代表当前在线的客户端客户端socket包装为EndpointClient ，有onClientAddDel 代表客户端上线掉线事件，有onReceive代表所有客户端的收包事件，clients由于是异步的多线程访问就要涉及多线程管控所以使用lock ，服务端有sendTo() 毫无疑问这也是通过调用特定的clients来做的。关于close() 网上说socket最好不要直接close ，但是我这里就是简单粗暴的直接close要不然系统底层socket迟迟不释放服务端再开的时候端口冲突，制造各种问题，何况都close了客户端也会断开连接还会管你数据报文有无接收完整吗，所以不要人云亦云，遇到问题了再说，结合自己的理解。

然后关于多报文格式的一些补充

这部分其实是后来写了更新上来的：从我们的报文数据定义为 Telegram_Base 就可看的出来，本意是想做成可扩展的，不要管Telegram_Base是什么往后面看你就明白了。并且硬性需求因为不止一个场景使用我们不可能把报文格式固定住肯定得做成可扩展的。为此我们使用模板编程方式，好久没有使用模板编程了，得复习下记得上次使用还是《angularjs和ajax的结合使用（四）》里面的 DownloadSomething<T> ，其实我也是半瓢水，哈哈哈。

MsgServer 和 MsgClient 初始化的时候传入不同的泛型，则Receive的时候根据不同的泛型解数据包进而触发 onReceive 事件，你要问为什么就想到这么做可以呢，一句话自然使然水到渠成，善于观察借鉴。我上面的处理方式，包括用过的supersocket 也都是这种处理方式。好改造开始，server和client都变成这样，总之一切都是围绕泛型报文展开的，包括send 和receive都要变成send<T> receive<T> 这样的

整体的类图结构：

左边代表服务端，初始化示例的时候可以通过泛型类方式绑定Telegram_xxx等不同的报文处理过程，服务端有个全局的缓冲字节receivedbuffer ，收到字节时调用不同报文对应的解析器解析，每当解析到完整包的时候通过action回调执行对应的用户自定义代码，右边是客户端发送时自然跟服务端报文对应。大概就这样，整体设计跟上面所描述一致，不太擅长画图也不知道表述清楚没有。

以下是服务端代码

  1 //消息服务器
  2 public class MsgServer<T> where T : Telegram_Base
  3 {
  4 
  5 
  6     Socket serverSocket;
  7     public Action<List<string>> onClientAddDel;
  8     public Action<T> onReceive;
  9     bool _isRunning = false;
 10 
 11     int port;
 12 
 13     static List<EndpointClient<T>> clients;
 14 
 15     public bool isRunning { get { return _isRunning; } }
 16     public MsgServer(int _port)
 17     {
 18         this.port = _port;
 19 
 20         clients = new List<EndpointClient<T>>();
 21 
 22     }
 23     
 24     public void Start()
 25     {
 26         try
 27         {
 28             //any 就决定了 ip地址格式是v4
 29             //IPEndPoint endPoint = new IPEndPoint(IPAddress.Any, 7654);
 30             //socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
 31             IPEndPoint endPoint = new IPEndPoint(IPAddress.Any, port);
 32             serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
 33             serverSocket.Bind(endPoint);
 34             serverSocket.Listen(port);
 35 
 36             serverSocket.BeginAccept(new AsyncCallback(AcceptCallback), serverSocket);
 37 
 38             _isRunning = true;
 39 
 40             Logger.GetInstance().AppendMessage(LogLevel.Info, "消息服务启动完成");
 41         }
 42         catch (Exception ex)
 43         {
 44             _isRunning = false;
 45             serverSocket = null;
 46 
 47             Logger.GetInstance().AppendMessage(LogLevel.Error, ex.Message);
 48         }
 49 
 50     }
 51 
 52     public void Stop()
 53     {
 54         for (int i = 0; i < clients.Count; i++)
 55         {
 56             clients[i].Close();
 57         }
 58         ClientAddDelGetList(null, EndPointClientsChangeType.ClearAll);
 59         serverSocket.Close();
 60         _isRunning = false;
 61     }
 62 
 63 
 64     public void SendTo(T tel)
 65     {
 66         try
 67         {
 68             if (string.IsNullOrEmpty(tel.remoteIPPort))//发送到所有
 69             {
 70                 for (int i = 0; i < clients.Count; i++)
 71                 {
 72                     clients[i].Send(tel);
 73                 }
 74             }
 75             else
 76             {
 77                 for (int i = 0; i < clients.Count; i++)//发送到某个客户端
 78                 {
 79                     if (clients[i].remoteIPPort == tel.remoteIPPort)
 80                     {
 81                         clients[i].Send(tel);
 82                         break;
 83                     }
 84                 }
 85             }
 86 
 87         }
 88         catch (Exception ex)
 89         {
 90             Logger.GetInstance().AppendMessage(LogLevel.Error, ex.Message);
 91         }
 92     }
 93 
 94     //新增与删除客户端 秉持原子操作
 95     List<string> ClientAddDelGetList(EndpointClient<T> cli, EndPointClientsChangeType changeType)
 96     {
 97         //异步同时有新客户端上线 与下线 不进行资源互斥访问会报错
 98         lock (this)
 99         {
100             if (changeType == EndPointClientsChangeType.Add)
101             {
102                 clients.Add(cli);
103             }
104             else if (changeType == EndPointClientsChangeType.Del)
105             {
106                 var beRemoveClient = clients.First(r => r.remoteIPPort == cli.remoteIPPort);
107                 if (beRemoveClient != null)
108                     clients.Remove(beRemoveClient);
109             }
110             else if (changeType == EndPointClientsChangeType.ClearAll)
111             {
112                 clients.Clear();
113             }
114             else if (changeType == EndPointClientsChangeType.GetAll)
115             {
116                 List<string> onLines = new List<string>(clients.Count);
117                 for (int i = 0; i < clients.Count; i++)
118                 {
119                     onLines.Add(clients[i].remoteIPPort);
120                 }
121                 return onLines;
122             }
123             else
124             {
125                 return null;
126             }
127         }
128         return null;
129     }
130     //异步监听客户端 有客户端到来时的回调
131     private void AcceptCallback(IAsyncResult iar)
132     {
133         //server端一直在receive 能够感知到客户端掉线 （连上后 关闭客户端 server立即有错误爆出）
134         //但是同情况 关闭server端 客户端无错误爆出 直到点发送 才有错误爆出
135         //由此得出 处于receive才会有掉线感知  ，send时发现发不出去自然也会有感知 跟人的正常思维理解是一样的
136         //虽然tcp是所谓的长连接 ，通过反复测试  ->但是双方相互都处在一个静止状态 是无法 确定在不在的  
137         //连上后平常的情况下 并没有数据流通 的 ,双方只是一个状态的保持而已。
138         //这也是为什么 好多服务 客户端 程序 都有个心跳机制（由此我们可以想到继续完善 弄个客户端列表 心跳超时的剔除列表 正常发消息send 或receive 异常的剔除列表 删除clientSocket
139         //其实非要说吧 一般情况 服务端一直在receive 不用心跳其实也是可以的（客户端可能是真的需要
140         //tcp底层就已经有了一个判断对方在不在的机制 ， 对方直接关程序 结束进程 这些 只要tcp在receive就立即能够感知 所以说心跳 用不用看情况吧
141 
142         //tcp 不会丢包 哪怕是连接建立了   你还没开始receive   对方却先发了，
143         //对方只要是发了的数据 都由操作系统像个缓存样给你放那的 不会掉 你再隔10秒开始receive都能rec的到
144 
145         //tcp甚至在拔掉网线 再重新插上 都可以保证数据一致性
146         //tcp的包顺序能够保证 先发的先到
147 
148         //nures代码中那些beginreceivexxx  异步receive的核心机制就是 ，假定数据到的时候把数据保存到xxx数组
149         //真正endreceive的时候 其实数据已经接收 处理完成了
150         try
151         {
152 
153             //处理完当前accept
154             Socket currentSocket = serverSocket.EndAccept(iar);
155 
156             EndpointClient<T> client = new EndpointClient<T>(currentSocket);
157 
158             //新增客户端
159             ClientAddDelGetList(client, EndPointClientsChangeType.Add);
160 
161             if (onClientAddDel != null)
162             {
163                 List<string> onlines = ClientAddDelGetList(null, EndPointClientsChangeType.GetAll);
164                 onClientAddDel(onlines);
165 
166                 //客户端异常掉线
167                 client.onClientDel = new Action<string>((_remoteIPPort) =>
168                 {
169                     ClientAddDelGetList(new EndpointClient<T>() { remoteIPPort = _remoteIPPort }, EndPointClientsChangeType.Del);
170 
171                     List<string> onlines2 = ClientAddDelGetList(null, EndPointClientsChangeType.GetAll);
172                     onClientAddDel(onlines2);
173                 });
174             }
175 
176 
177             Logger.GetInstance().AppendMessage(LogLevel.Debug, string.Format("new client ->{0}", currentSocket.RemoteEndPoint.ToString()));
178             //Console.WriteLine(string.Format("new client ->{0}", currentSocket.RemoteEndPoint.ToString()));
179 
180             //currentSocket.Close();
181             //Application.Exit();
182 
183             //Thread.Sleep(1000 * 10);
184             client.onReceive += this.onReceive;
185 
186             client.BeginReceive();
187 
188 
189             //立即开始accept新的客户端
190             if (isRunning == true && serverSocket != null)
191                 serverSocket.BeginAccept(AcceptCallback, serverSocket);
192             //beginAccept 最开始的方法可以不一样 ，但最终肯定是一个不断accept的闭环过程
193             //整个过程就像个导流样 ，最开始用异步导流到一个固定的点 然后让其循环源源不断运转
194 
195         }
196         catch (Exception ex)
197         {
198             Logger.GetInstance().AppendMessage(LogLevel.Error, ex.Message);
199             Console.WriteLine(ex.Message);
200         }
201 
202     }
203 
204 
205 }

EndpointClient终端代码代表客户端的对口人，他的onReceive 等资源从服务端继承，如果服务端想给某个特定客户端发数据则会调用他们中的某一个毫无疑问这是通过remoteIPport来判断的，这些都是编写基本socket结构轻车熟路的老套路

以下EndpointClient代码

  1 public class EndpointClient<T> where T : Telegram_Base
  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 
 15     public Action<T> onReceive;
 16     public Action<string> onClientDel;
 17 
 18     public EndpointClient()
 19     {
 20 
 21     }
 22     public EndpointClient(Socket _socket)
 23     {
 24         this.remoteIPPort = _socket.RemoteEndPoint.ToString();
 25         workingSocket = _socket;
 26     }
 27 
 28     public void Send(T tel)
 29     {
 30         //try
 31         //{
 32         if (workingSocket == null)
 33         {
 34             Console.WriteLine("未初始化的EndpointClient");
 35             return;
 36         }
 37         if (tel is Telegram_Schedule)
 38         {
 39             Telegram_Schedule telBeSend = tel as Telegram_Schedule;
 40             if (telBeSend.dataBytes.Length != telBeSend.dataLen)
 41             {
 42                 Console.WriteLine("尝试发送数据长度格式错误的报文");
 43                 return;
 44             }
 45 
 46             byte[] sendBytesHeader = telBeSend.dataBytesHeader;
 47             byte[] sendbytes = telBeSend.dataBytes;
 48 
 49             //数据超过缓冲区长度 会导致无法拆包
 50             if (sendbytes.Length <= receiveBufferLenMax)
 51             {
 52                 workingSocket.BeginSend(sendBytesHeader, 0, sendBytesHeader.Length, 0, null, null);
 53                 workingSocket.BeginSend(sendbytes, 0, sendbytes.Length, 0, null, null
 54                 //    new AsyncCallback((iar) =>
 55                 //{
 56                 //    EndpointClient endClient = (EndpointClient)iar.AsyncState;
 57                 //    endClient.workingSocket.EndSend(iar);
 58                 //}),
 59                 //this
 60                 );
 61             }
 62             else
 63             {
 64                 Console.WriteLine("发送到调度客户端的数据超过缓冲区长度");
 65                 throw new Exception("发送到调度客户端的数据超过缓冲区长度");
 66             }
 67 
 68 
 69         }
 70         else if (tel is Telegram_SDBMsg)
 71         {
 72             Telegram_SDBMsg telBesendSDB = tel as Telegram_SDBMsg;
 73             if (telBesendSDB.dataBytes.Length != telBesendSDB.dataLen)
 74             {
 75                 Console.WriteLine("尝试发送数据长度格式错误的报文");
 76                 return;
 77             }
 78             byte[] sendBytesHeader = telBesendSDB.dataBytesHeader;
 79             byte[] sendbytes = telBesendSDB.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         //catch (Exception ex)
 98         //{
 99 
100         //    Console.WriteLine(ex.Message);
101         //    throw ex;
102         //}
103     }
104 
105     public void BeginReceive()
106     {
107         if (workingSocket == null)
108         {
109             Console.WriteLine("未初始化的EndpointClient");
110             return;
111         }
112 
113         receiveBufferLen = 0;
114         workingSocket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax, SocketFlags.None,
115             ReceiveCallback,
116         //    new AsyncCallback((IAsyncResult iar) => {
117         //    EndpointClient cli = (EndpointClient)iar.AsyncState;
118         //    int reds = cli.client.EndReceive(iar);
119         //}),
120         this);
121     }
122     private void ReceiveCallback(IAsyncResult iar)
123     {
124         try
125         {
126             EndpointClient<T> cli = (EndpointClient<T>)iar.AsyncState;
127             Socket socket = cli.workingSocket;
128             int reads = socket.EndReceive(iar);
129 
130             if (reads > 0)
131             {
132 
133                 for (int i = 0; i < reads; i++)
134                 {
135                     receiveBuffer.Add(onceReadDatas[i]);
136                 }
137 
138                 //具体填充了多少看返回值 此时 数据已经在buffer中了
139                 receiveBufferLen += reads;
140                 //加完了后解析 阻塞式处理 结束后开启新的接收
141                 SloveTelData();
142 
143                 if (receiveBufferLenMax - receiveBufferLen > 0)
144                 {
145                     //接收完了 继续beginreceive 开启异步的下次接收 (如果缓冲区有残留数据 则接收长度变短 ，没接收到的让其留在socket不会丢失 下次接收)
146                     socket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax - receiveBufferLen, SocketFlags.None, ReceiveCallback, this);
147                 }
148                 else//阻塞式处理都完成一遍了 都还没清理出任何缓冲区空间 毫无疑问 整体运转机制已经挂了 不用beginreceive下一次了
149                 {
150                     Close();
151                     //移除自己
152                     if (onClientDel != null)
153                     {
154                         onClientDel(remoteIPPort);
155                     }
156                     Logger.GetInstance().AppendMessage(LogLevel.Error, "服务端接口解析数据出现异常");
157                     //Console.WriteLine("服务端接口解析数据出现异常");
158                     //throw new Exception("服务端接口解析数据出现异常");
159                 }
160             }
161             else//reads==0 客户端已关闭
162             {
163                 Close();
164                 //移除自己
165                 if (onClientDel != null)
166                 {
167                     onClientDel(remoteIPPort);
168                 }
169             }
170         }
171         catch (Exception ex)
172         {
173             Close();
174             //移除自己
175             if (onClientDel != null)
176             {
177                 onClientDel(remoteIPPort);
178             }
179             Logger.GetInstance().AppendMessage(LogLevel.Error, "ReceiveCallback Error" + ex.Message);
180             //Console.WriteLine("ReceiveCallback Error");
181             //Console.WriteLine(ex.Message);
182         }
183 
184     }
185     void SloveTelData()
186     {
187         //进行数据解析
188 
189 
190         if (typeof(T) == typeof(Telegram_Schedule))
191         {
192             SloveTelDataUtil slo = new SloveTelDataUtil();
193             List<Telegram_Schedule> dataEntitys = slo.Slove_Telegram_Schedule(receiveBuffer, receiveBufferLen, this.remoteIPPort);
194             //buffer已经被处理一遍了 使用新的长度
195             receiveBufferLen = receiveBuffer.Count;
196             //解析出的每一个对象都触发 onreceive
197             for (int i = 0; i < dataEntitys.Count; i++)
198             {
199                 if (onReceive != null)
200                     onReceive(dataEntitys[i] as T);
201             }
202         }
203         else if (typeof(T) == typeof(Telegram_SDBMsg))
204         {
205             SloveTelSDBMsgUtil sloSDB = new SloveTelSDBMsgUtil();
206             List<Telegram_SDBMsg> dataEntitys = sloSDB.Slove_Telegram_SDB(receiveBuffer, receiveBufferLen, this.remoteIPPort);
207             receiveBufferLen = receiveBuffer.Count;
208             //解析出的每一个对象都触发 onreceive
209             for (int i = 0; i < dataEntitys.Count; i++)
210             {
211                 if (onReceive != null)
212                     onReceive(dataEntitys[i] as T);
213             }
214         }
215 
216     }
217 
218 
219     public void Close()
220     {
221         try
222         {
223             receiveBuffer.Clear();
224             receiveBufferLen = 0;
225             if (workingSocket != null && workingSocket.Connected)
226                 workingSocket.Close();
227         }
228         catch (Exception ex)
229         {
230             Console.WriteLine(ex.Message);
231         }
232 
233     }
234 }

数据拆包与封包粘包处理

上面的代码可以看到数据包处理都在receiveCallback里 SloveTelData，也是通用的套路，解析到完整的包后从缓冲区移除解析多少个包触发多少次事件，残余数据留在缓冲区然后继续开始新的beginReceive往缓冲区加。在异步机制中到达endReceive的时候数据已经在缓冲区里了，这个自不用多说噻。数据包和粘包逻辑在公共类库里供客户端服务端共同调用。

以下是Telegram_schedule数据包的粘包处理逻辑，当然看代码你知道还有另外一个报文处理逻辑我不列举了。

  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         buffer = _buffer;
 17         bufferLen = _bufferLen;
 18         bufferIndex = 0;
 19         remoteIPPort = _remoteIPPort;
 20 
 21         //小于最小长度 直接返回
 22         if (bufferLen < 12)
 23             return solveList;
 24 
 25         //进行数据解析
 26         bool anaysisOK = false;
 27         while (anaysisOK=AnaysisData_Schedule()==true)//直到解析的不能解析为止
 28         {                
 29         }
 30         return solveList;
 31     }
 32 
 33     public bool AnaysisData_Schedule()
 34     {
 35         if (bufferLen - bufferIndex < GlobalSymbol.Headerlen)
 36             return false;
 37 
 38         //解析出一个数据对象
 39         Telegram_Schedule telObj = new Telegram_Schedule();
 40 
 41         //必定是大于最小数据大小的
 42         telObj.dataBytesHeader = new byte[GlobalSymbol.Headerlen];
 43         buffer.CopyTo(bufferIndex, telObj.dataBytesHeader, 0, GlobalSymbol.Headerlen);
 44 
 45         byte[] btsHeader = new byte[4];
 46         byte[] btsCommand = new byte[4];
 47         byte[] btsLen = new byte[4];
 48 
 49         btsHeader[0] = buffer[bufferIndex];
 50         btsHeader[1] = buffer[bufferIndex+1];
 51         btsHeader[2] = buffer[bufferIndex+2];
 52         btsHeader[3] = buffer[bufferIndex+3];
 53 
 54         bufferIndex += 4;
 55 
 56         btsCommand[0] = buffer[bufferIndex];
 57         btsCommand[1] = buffer[bufferIndex + 1];
 58         btsCommand[2] = buffer[bufferIndex + 2];
 59         btsCommand[3] = buffer[bufferIndex + 3];
 60 
 61         bufferIndex += 4;
 62 
 63         btsLen[0] = buffer[bufferIndex];
 64         btsLen[1] = buffer[bufferIndex + 1];
 65         btsLen[2] = buffer[bufferIndex + 2];
 66         btsLen[3] = buffer[bufferIndex + 3];
 67 
 68         bufferIndex += 4;
 69 
 70         
 71 
 72         int dataLen = BitConverter.ToInt32(btsLen, 0);
 73         telObj.head = BitConverter.ToUInt32(btsHeader, 0);
 74         telObj.command = BitConverter.ToInt32(btsCommand, 0);
 75         telObj.remoteIPPort = remoteIPPort;
 76 
 77         if(dataLen>0)
 78         {
 79             //数据区小于得到的数据长度 说明数据部分还没接收到 不删除缓冲区 不做任何处理
 80             //下次来了连着头一起解析
 81             if (bufferLen - GlobalSymbol.Headerlen < dataLen)
 82             {
 83 
 84                 bufferIndex -= 12;//
 85 
 86 
 87                 return false;
 88 
 89             }
 90             else
 91             {
 92 
 93                 telObj.dataLen = dataLen;
 94                 telObj.dataBytes = new byte[dataLen];
 95                 buffer.CopyTo(bufferIndex, telObj.dataBytes, 0, dataLen);
 96                 
 97                 solveList.Add(telObj);
 98                 //bufferIndex += dataLen;
 99 
100                 //解析成功一次 移除已解析的
101                 for (int i = 0; i < GlobalSymbol.Headerlen+dataLen; i++)
102                 {
103                     buffer.RemoveAt(0);
104                 }
105                 bufferIndex = 0;
106                 bufferLen = buffer.Count;
107                 return true;
108             }
109         }
110         else
111         {
112             
113             telObj.dataLen = 0;
114             solveList.Add(telObj);
115             //bufferIndex += 0;
116             //解析成功一次 移除已解析的
117             for (int i = 0; i < GlobalSymbol.Headerlen; i++)
118             {
119                 buffer.RemoveAt(0);
120             }
121             //解析成功一次因为移除了缓冲区 bufferIndex置0
122             bufferIndex = 0;
123             bufferLen = buffer.Count;
124             return true;
125         }
126 
127     }
128 
129 }

我们看到用到的数据包对象是Telegram_Schedule ，中间保存有报文数据，数据发送的目标等信息。

以下是数据包结构代码

 1 public class Telegram_Base
 2 {
 3     public string remoteIPPort { get; set; }
 4     
 5     //头部内容的序列化
 6     public byte[] dataBytesHeader { get; set; }
 7 
 8     //数据内容
 9     public byte[] dataBytes { get; set; }
10 
11     public int headerLen { get; set; }
12     //数据长度 4字节
13     public int dataLen { get; set; }
14 
15     public string jsonStr { get; set; }
16     virtual public void SerialToBytes()
17     {
18 
19     }
20 
21     virtual public void SloveToTel()
22     {
23 
24     }
25 
26 }
27 
28 
29 public class Telegram_Schedule:Telegram_Base
30 {
31     
32     //头部标识 4字节
33     public UInt32 head { get; set; }
34     //命令对应枚举的 int 4字节
35     public int command { get; set; }
36    
37 
38     override public void SerialToBytes()
39     {
40         //有字符串数据 但是待发送字节是空
41         if ((string.IsNullOrEmpty(jsonStr) == false ))//&& (dataBytes==null || dataBytes.Length==0)
42         {
43             dataBytes = Encoding.UTF8.GetBytes(jsonStr);
44             dataLen = dataBytes.Length;
45             dataBytesHeader = new byte[GlobalSymbol.Headerlen];
46           
47             head = GlobalSymbol.HeaderSymbol;
48             
49             byte[] btsHeader = BitConverter.GetBytes(head);
50             byte[] btsCommand = BitConverter.GetBytes(command);
51             byte[] btsLen = BitConverter.GetBytes(dataLen);
52 
53             Array.Copy(btsHeader, 0, dataBytesHeader, 0, 4);
54             Array.Copy(btsCommand, 0, dataBytesHeader, 4, 4);
55             Array.Copy(btsLen, 0, dataBytesHeader, 8, 4);
56 
57         }
58         else if((string.IsNullOrEmpty(jsonStr) == true )&& (dataBytes==null || dataBytes.Length==0)){
59             dataLen = 0;
60             dataBytes = new byte[0];
61 
62             dataBytesHeader = new byte[GlobalSymbol.Headerlen];
63 
64             head = GlobalSymbol.HeaderSymbol;
65 
66             byte[] btsHeader = BitConverter.GetBytes(head);
67             byte[] btsCommand = BitConverter.GetBytes(command);
68             byte[] btsLen = BitConverter.GetBytes(dataLen);
69 
70             Array.Copy(btsHeader, 0, dataBytesHeader, 0, 4);
71             Array.Copy(btsCommand, 0, dataBytesHeader, 4, 4);
72             Array.Copy(btsLen, 0, dataBytesHeader, 8, 4);
73         }
74     }
75 
76     override public void SloveToTel()
77     {
78         //只解析字符串数据部分 ，header 和len 在接收之初就已解析
79         try
80         {
81             if (this.dataBytes != null && this.dataBytes.Length > 0)
82                 this.jsonStr = Encoding.UTF8.GetString(this.dataBytes);
83         }
84         catch (Exception ex)
85         {
86             Logger.GetInstance().AppendMessage(LogLevel.Error, "data部分字符串解析出错 " + ex.Message);
87         }
88     }
89 
90 }

客户端代码

最后是客户端，有了上面的结构，客户端就不足为谈了，稍微了解socket的人都熟知套路的基本跟EndpointClient一致

  1 public class MsgClient<T> where T : Telegram_Base
  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 
 21     //收一个包时触发
 22     public Action<T> onReceive;
 23     //与服务端断链时触发
 24     public Action<string> onClientDel;
 25 
 26 
 27     public bool isConnected { get { return _isConnected; } }
 28     public MsgClient(string _serverIP, int _port)
 29     {
 30         serverIP = _serverIP;
 31         serverPort = _port;
 32         _isConnected = false;
 33     }
 34 
 35     public void Connect()
 36     {
 37         try
 38         {
 39             workingSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.IP);
 40             IPEndPoint ipport = new IPEndPoint(IPAddress.Parse(serverIP), serverPort);
 41             workingSocket.Connect(ipport);
 42 
 43             localIPPort = workingSocket.LocalEndPoint.ToString();
 44             _isConnected = true;
 45             BeginReceive();
 46         }
 47         catch (Exception ex)
 48         {
 49             workingSocket = null;
 50             _isConnected = false;
 51 
 52             Logger.GetInstance().AppendMessage(LogLevel.Error,"连接到服务端出错："+ ex.Message);
 53         }
 54 
 55     }
 56 
 57 
 58 
 59 
 60     public void Send(T tel)
 61     {
 62         try
 63         {
 64             if (_isConnected == false)
 65             {
 66                 Console.WriteLine("未连接到服务器");
 67                 return;
 68             }
 69             if (tel is Telegram_Schedule)
 70             {
 71                 Telegram_Schedule telBeSend = tel as Telegram_Schedule;
 72                 if (telBeSend.dataBytes.Length != telBeSend.dataLen)
 73                 {
 74                     Console.WriteLine("尝试发送数据长度格式错误的报文");
 75                     return;
 76                 }
 77                 byte[] sendBytesHeader = telBeSend.dataBytesHeader;
 78                 byte[] sendbytes = telBeSend.dataBytes;
 79 
 80                 //数据超过缓冲区长度 会导致无法拆包
 81                 if (sendbytes.Length <= receiveBufferLenMax)
 82                 {
 83                     workingSocket.BeginSend(sendBytesHeader, 0, sendBytesHeader.Length, 0, null, null);
 84                     workingSocket.BeginSend(sendbytes, 0, sendbytes.Length, 0, null, null
 85 
 86                     );
 87                 }
 88                 else
 89                 {
 90                     Logger.GetInstance().AppendMessage(LogLevel.Error, "发送到调度客户端的数据超过缓冲区长度");
 91                     throw new Exception("发送到调度客户端的数据超过缓冲区长度");
 92                 }
 93 
 94 
 95             }
 96             else if (tel is Telegram_SDBMsg)
 97             {
 98                 Telegram_SDBMsg telBesendSDB = tel as Telegram_SDBMsg;
 99                 if (telBesendSDB.dataBytes.Length != telBesendSDB.dataLen)
100                 {
101                     Console.WriteLine("尝试发送数据长度格式错误的报文");
102                     return;
103                 }
104                 byte[] sendBytesHeader = telBesendSDB.dataBytesHeader;
105                 byte[] sendbytes = telBesendSDB.dataBytes;
106 
107                 //数据超过缓冲区长度 会导致无法拆包
108                 if (sendbytes.Length <= receiveBufferLenMax)
109                 {
110                     workingSocket.BeginSend(sendBytesHeader, 0, sendBytesHeader.Length, 0, null, null);
111                     workingSocket.BeginSend(sendbytes, 0, sendbytes.Length, 0, null, null);
112                 }
113                 else
114                 {
115                     Logger.GetInstance().AppendMessage(LogLevel.Error, "发送到调度客户端的数据超过缓冲区长度");
116                     throw new Exception("发送到调度客户端的数据超过缓冲区长度");
117                 }
118             }
119 
120         }
121         catch (Exception ex)
122         {
123             Logger.GetInstance().AppendMessage(LogLevel.Error, ex.Message);
124             Close();
125             //Console.WriteLine(ex.Message);
126             //throw ex;
127         }
128     }
129 
130     public void BeginReceive()
131     {
132         receiveBufferLen = 0;
133         workingSocket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax, SocketFlags.None,
134             ReceiveCallback,
135 
136         this);
137     }
138     private void ReceiveCallback(IAsyncResult iar)
139     {
140         try
141         {
142             MsgClient<T> cli = (MsgClient<T>)iar.AsyncState;
143             Socket socket = cli.workingSocket;
144             int reads = socket.EndReceive(iar);
145 
146             if (reads > 0)
147             {
148 
149                 for (int i = 0; i < reads; i++)
150                 {
151                     receiveBuffer.Add(onceReadDatas[i]);
152                 }
153 
154                 //具体填充了多少看返回值 此时 数据已经在buffer中了
155 
156                 receiveBufferLen += reads;
157 
158                 //加完了后解析 阻塞式处理 结束后开启新的接收
159                 SloveTelData();
160 
161 
162 
163                 if (receiveBufferLenMax - receiveBufferLen > 0)
164                 {
165                     //接收完了 继续beginreceive 开启异步的下次接收 (如果缓冲区有残留数据 则接收长度变短 ，没接收到的让其留在socket不会丢失 下次接收)
166                     socket.BeginReceive(onceReadDatas, 0, receiveBufferLenMax - receiveBufferLen, SocketFlags.None, ReceiveCallback, this);
167                 }
168                 else//阻塞式处理都完成一遍了 都还没清理出任何缓冲区空间 毫无疑问 整体运转机制已经挂了 不用beginreceive下一次了
169                 {
170                     Close();
171 
172                     Console.WriteLine("服务端接口解析数据出现异常");
173                     throw new Exception("服务端接口解析数据出现异常");
174                 }
175             }
176             else//reads==0客户端已关闭
177             {
178                 Close();
179             }
180         }
181         catch (Exception ex)
182         {
183             Close();
184 
185             Console.WriteLine("ReceiveCallback Error");
186             Console.WriteLine(ex.Message);
187         }
188 
189     }
190     private void SloveTelData()
191     {
192 
193         //进行数据解析
194         
195 
196         if (typeof(T) == typeof(Telegram_Schedule))
197         {
198             SloveTelDataUtil slo = new SloveTelDataUtil();
199             List<Telegram_Schedule> dataEntitys = slo.Slove_Telegram_Schedule(receiveBuffer, receiveBufferLen, serverIP + ":" + serverPort.ToString());
200             //buffer已经被处理一遍了 使用新的长度
201             receiveBufferLen = receiveBuffer.Count;
202             //解析出的每一个对象都触发 onreceive
203             for (int i = 0; i < dataEntitys.Count; i++)
204             {
205                 if (onReceive != null)
206                     onReceive(dataEntitys[i] as T);
207             }
208         }
209         else if (typeof(T) == typeof(Telegram_SDBMsg))
210         {
211             SloveTelSDBMsgUtil sloSDB = new SloveTelSDBMsgUtil();
212             List<Telegram_SDBMsg> dataEntitys = sloSDB.Slove_Telegram_SDB(receiveBuffer, receiveBufferLen, serverIP + ":" + serverPort.ToString());
213             receiveBufferLen = receiveBuffer.Count;
214             //解析出的每一个对象都触发 onreceive
215             for (int i = 0; i < dataEntitys.Count; i++)
216             {
217                 if (onReceive != null)
218                     onReceive(dataEntitys[i] as T);
219             }
220         }
221 
222     }
223 
224 
225     public void Close()
226     {
227         try
228         {
229             _isConnected = false;
230 
231             receiveBuffer.Clear();
232             receiveBufferLen = 0;
233             if (workingSocket != null && workingSocket.Connected)
234                 workingSocket.Close();
235         }
236         catch (Exception ex)
237         {
238             Console.WriteLine(ex.Message);
239         }
240 
241     }
242 
243 }

服务端调用

构建一个winform基本项目

 1 List<string> clients;
 2 TCPListener server2;
 3 private void button1_Click(object sender, EventArgs e)
 4 {
 5     server = new MsgServer<Telegram_Schedule>(int.Parse(tbx_port.Text));
 6 
 7     server.Start();
 8     if (server.isRunning == true)
 9     {
10         button1.Enabled = false;
11 
12         server.onReceive += new Action<Telegram_Base>(
13         (tel) =>
14         {
15             this.BeginInvoke(new Action(() =>
16             {
17                 if (tel is Telegram_Schedule)
18                 {
19                     Telegram_Schedule ts = tel as Telegram_Schedule;
20                     ts.SloveToTel();
21                     //Console.WriteLine(string.Format("commandType:{0}", ((ScheduleTelCommandType)ts.command).ToString()));
22 
23                     textBox1.Text += ts.remoteIPPort + ">" + ts.jsonStr + "\r\n";
24 
25                     //数据回发测试
26                     string fromip = ts.remoteIPPort;
27                     string srcMsg = ts.jsonStr;
28                     string fromServerMsg = ts.jsonStr + " -from server";
29                     ts.jsonStr = fromServerMsg;
30 
31 
32                     //如果消息里有指向信息 则转送到对应的客户端
33                     if (clients != null)
34                     {
35                         string to = null;
36                         for (int i = 0; i < clients.Count; i++)
37                         {
38                             if (srcMsg.Contains(clients[i]))
39                             {
40                                 to = clients[i];
41                                 break;
42                             }
43                         }
44 
45                         if (to != null)
46                         {
47                             ts.remoteIPPort = to;
48                             string toMsg;
49                             //toMsg= srcMsg.Replace(to, "");
50                             toMsg = srcMsg.Replace(to, fromip);
51                             ts.jsonStr = toMsg;
52                             ts.SerialToBytes();
53 
54                             server.SendTo(ts);
55                         }
56                         else
57                         {
58                             ts.SerialToBytes();
59                             server.SendTo(ts);
60                         }
61                     }
62                 }
63             }));
64 
65         }
66         );
67 
68         server.onClientAddDel += new Action<List<string>>((onlines) =>
69         {
70             this.BeginInvoke(
71                 new Action(() =>
72                 {
73                     clients = onlines;
74                     listBox1.Items.Clear();
75 
76                     for (int i = 0; i < onlines.Count; i++)
77                     {
78                         listBox1.Items.Add(onlines[i]);
79                     }
80                 }));
81         });
82     }
83 }

客户端调用

 1 MsgClient<Telegram_Schedule> client;
 2 TCPClient client2;
 3 private void btn_start_Click(object sender, EventArgs e)
 4 {
 5     client = new MsgClient<Telegram_Schedule>(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 + "\r\n";
22 
23                 }));
24         });
25     }
26 
27 }
28      
29 
30 private void btn_send_Click(object sender, EventArgs e)
31 {
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 }

实现效果

可以多客户端连接互相自由发送消息，服务端可以编写转发规则代码，那些什么棋牌啊互动白板以及其他类似的应用就可以基于此之上发挥想象了。标题叫一个简易socket通信结构，一路走下来你看整个结构其实也并不是那么简易，断断续续调了几周解决了很多问题。完美的支持报文格式切换，这里只是拿聊天功能做个示例，但是其实这已经是一个由我编写出来然后测试并成功商业应用了的部分哈，可靠稳定你值得拥有。都知道流行GitHub 但是本人并不习惯把代码放GitHub上见谅。

出处：https://www.cnblogs.com/assassinx/p/17154167.html

栏目列表