分享一下，项目中写的简单socket程序，不同方式的版本，这是个异步基于IOCP实现高性能的处理方式。IOCP就不多说了，高性能的完成端口，可以实现套接字对象的复用，降低开销，且基于端口共享性能据说很高，因此把之前的项目做个改动。下面来个代码走读：
功能介绍：
此socket程序类似一个交换机，对下统一采用报文来交换数据，什么心跳、回执等各式各样十几种类，他负责接收各个client发送来的消息，向各个client回发确认帧、发送对设备任务请求等工作；对上，向数据库存入设备状态等数据，将应用层的任务，组织合适的报文，下发给client。
他所处的角色和位置：有时候，接收到client的请求后直接转发给BS（本质就是存入数据库啦）;有时候，接收到client的请求后 思考 一下，向应用层拉取数据，装配为 恰当的报文 后，回送给对应的client；轮询BS提供的业务接口，发现有对设备请求的任务后，装配为恰当的格式后，发送给对应的client。
就像一个繁忙的机械手，有时对于收到的报文，仅仅是左手接，逻辑处理后，右手丢回去，必须满足，从哪里接的、需送回给正确的client；有时候，只是充当一个转发器，收到的内容不做任何逻辑处理，直接丢给第三方处理。虽然程序都已经运行多时，都没有想好一个恰当的名字来命名程序，他更像是一个桥、一个转换器。
下面是主进程需要声明的符号；收件箱存入接收的消息，不做任何处理；会有相应的进程不间断的扫描收件队列，一旦发现内容，便开始处理，处理结果会存入对应的发件箱，同样有相应的进程不间断的处理，通常为发送至client端，或者发送至BS接口。
1 private readonly Config _config = InitConfig.ReadConfig();//配置信息 端口 IP 2 3 private static IDictionary<int, WcsEndpoint<Socket>> _wcsList = new Dictionary<int, WcsEndpoint<Socket>>();//记录有多少个端点连接进来 4 private const int BufferSize = 1024; 5 private static SocketTool _socketTool = new SocketTool();//完成分包、压缩、加密关键报文内容 6 private static MessageFactory _messageFactory = new MessageFactory();//用于构造不同的消息报文 7 private static TaskSendDown _taskStockIn = new TaskSendDown(_socketTool);//用于桥接浏览器，通知web端一些事情发生 8 private static ConcurrentQueue<ReceiveEntity> _receiveMailBox = new ConcurrentQueue<ReceiveEntity>();//收件箱 9 private static ConcurrentQueue<ReceiveEntity> _sendMailBox = new ConcurrentQueue<ReceiveEntity>();//发件箱 10 private static object _lockwcsList = new object();//同步锁 11 private static CachePool _sPoolyncDataPool = new CachePool();//扩展数据缓存
1 var ip = IPAddress.Parse(_config.LocalIp); 2 TcpListener.InstanceSocketTool = _socketTool; 3 TcpListener.LockwcsList = _lockwcsList; 4 TcpListener.WcsList = _wcsList; 5 var core = new TcpListener 6 {//将外部变量传递给核心处理类 7 MaxConnect = int.Parse(_config.MaxConnect), 8 ReceiveMailBox = _receiveMailBox, 9 SPoolyncDataPool = _sPoolyncDataPool, 10 SendMailBox = _sendMailBox, 11 TaskStockIn = _taskStockIn, 12 }; 13 CallbackUpdateStrip("启动wcs监听;");//同步UI 14 var receiveBoxHandler = new Thread(core.HandReceiveMailbox); //收件邮箱侦听 15 receiveBoxHandler.Start(); 16 17 var sendBoxHandler = new Thread(core.HandleSendMailbox); //发件箱侦听 18 sendBoxHandler.Start(); 19 core.Listen(new IPEndPoint(ip, int.Parse(_config.PortForWcs)));
核心socket实现，抄袭了别人的稍微改造一下，或者高端点重构了一下：
1 using System; 2 using System.Collections.Concurrent; 3 using System.Collections.Generic; 4 using System.Diagnostics; 5 using System.Drawing; 6 using System.Linq; 7 using System.Net; 8 using System.Net.Sockets; 9 using System.Runtime.InteropServices; 10 using System.Runtime.Remoting.Contexts; 11 using System.Text; 12 using System.Threading; 13 using Newtonsoft.Json; 14 using NovaMessageSwitch; 15 using NovaMessageSwitch.Bll; 16 using NovaMessageSwitch.message; 17 using NovaMessageSwitch.Model; 18 using NovaMessageSwitch.Tool; 19 using NovaMessageSwitch.Tool.DataCache; 20 using NovaMessageSwitch.Tool.Log; 21 22 /**************************************************************** 23 * 作者：wxy 24 * CLR版本：4.0.30319.42000 25 * 创建时间：2016/3/25 8:47:51 26 * 2016 27 * 描述说明： 28 * SocketAsyncEventArgs是异步的socket类，封装了IOCP，可以很方便实现non-blocking IO 29 非阻塞IO对server性能和吞吐量有很大好处 30 * 修改历史： 31 * IPAddress[] addressList = Dns.GetHostEntry(Environment.MachineName).AddressList; 32 new TcpListener().Listen(/new IPEndPoint(addressList[addressList.Length - 1]/, 9900)); 33 34 Console.ReadKey(); 35 36 * 37 *****************************************************************/ 38 namespace CompletePortB 39 { 40 public class TcpListener 41 { 42 private SocketAsyncEventArgs _args; 43 private Socket _listenerSocket; 44 private StringBuilder _buffers; 45 private const int BufferSize = 1024; 46 47 public TcpListener() 48 { 49 MaxConnect = 1000; 50 51 } 52 53 public int MaxConnect { get; set; } 54 public static object LockwcsList { get; set; } 55 public static IDictionary<int, WcsEndpoint<Socket>> WcsList { get; set; } 56 public TaskSendDown TaskStockIn { get; set; } 57 public CachePool SPoolyncDataPool { get; set; } 58 public ConcurrentQueue<ReceiveEntity> ReceiveMailBox { get; set; } 59 public ConcurrentQueue<ReceiveEntity> SendMailBox { get; set; } 60 61 public static SocketTool InstanceSocketTool { get; set; } 62 63 public void Listen(EndPoint e) 64 { 65 try 66 { 67 _buffers = new StringBuilder(); 68 _listenerSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); 69 _listenerSocket.Bind(e); 70 _listenerSocket.Listen(MaxConnect); 71 _args = new SocketAsyncEventArgs(); 72 _args.Completed += new EventHandler<SocketAsyncEventArgs>(ProcessAccept); 73 BeginAccept(_args); 74 Console.Read(); 75 } 76 catch (Exception ex) 77 { 78 79 } 80 } 81 82 private void BeginAccept(SocketAsyncEventArgs e) 83 { 84 e.AcceptSocket = null; 85 //异步操作完成，返回false 86 if (!_listenerSocket.AcceptAsync(e)) 87 ProcessAccept(_listenerSocket, e); 88 89 } 90 91 //异步操作完成时调用此方法 92 private void ProcessAccept(object sender, SocketAsyncEventArgs e) 93 { 94 var s = e.AcceptSocket; 95 e.AcceptSocket = null; 96 var args = new SocketAsyncEventArgs(); 97 //继续向下回掉函数 98 args.Completed += new EventHandler<SocketAsyncEventArgs>(OnIoCompleted); 99 //设定异步套接字方法的数据缓冲区 100 args.SetBuffer(new byte[BufferSize], 0, BufferSize); 101 args.AcceptSocket = s; 102 //IO操作同步完成，返回false,且不引发e参数的socketasynceventargs事件 103 //IO操作若被挂起，将返回true,且引发e参数的... 104 if (!s.ReceiveAsync(args)) 105 ProcessReceive(args); 106 BeginAccept(e); 107 } 108 void OnIoCompleted(object sender, SocketAsyncEventArgs e) 109 { 110 //获取最近使用此上下文对象执行的套接字操作类型 111 switch (e.LastOperation) 112 { 113 case SocketAsyncOperation.Receive: 114 this.ProcessReceive(e); 115 break; 116 case SocketAsyncOperation.Send: 117 this.ProcessSend(e); 118 break; 119 default: 120 throw new ArgumentException("The last operation completed on the socket was not a receive or send"); 121 } 122 } 123 124 private void ProcessSend(SocketAsyncEventArgs e) 125 { 126 if (e.SocketError == SocketError.Success) 127 { 128 //接收完毕开始下次接收 129 if (!e.AcceptSocket.ReceiveAsync(e)) 130 { 131 this.ProcessReceive(e); 132 } 133 } 134 else 135 { 136 137 } 138 139 } 140 141 142 /// <summary> 143 /// 维护wcs端点列表 144 /// </summary> 145 /// <param name="clientId"></param> 146 /// <param name="socket"></param> 147 private static void AddDictWcs(int clientId, Socket socket) 148 { 149 lock (LockwcsList) 150 { 151 if (WcsList.Keys.Contains(clientId)) 152 { 153 var endPoint = WcsList[clientId]; 154 endPoint.EndPoint = socket; 155 endPoint.RecentTimeOld = endPoint.RecentTime; 156 endPoint.RecentTime = DateTime.Now; 157 InstanceSocketTool.UpdateWcsDisplay(endPoint, UpdateUi.Post); 158 } 159 else 160 { 161 var newEndPoit = new WcsEndpoint<Socket> 162 { 163 RecentTime = DateTime.Now, 164 RecentTimeOld = null, 165 EndPoint = socket 166 }; 167 WcsList.Add(clientId, newEndPoit); 168 } 169 170 foreach (var wcs in WcsList.Where(wcs => (wcs.Value.RecentTime - DateTime.Now).Minutes > 2)) 171 { 172 WcsList.Remove(wcs.Key); 173 } 174 InstanceSocketTool.UpdateWcsDisplay(WcsList, UpdateUi.Post); 175 } 176 } 177 private string ReplyAckWcs(Socket socket, string oriSerial) 178 { 179 var socketTool = InstanceSocketTool; 180 var ackMessage = new MessageData<ContentReply> 181 { 182 infoType = 1, 183 content = new ContentReply(), 184 destination = DataFlowDirection.wcs.ToString(), 185 source = DataFlowDirection.wms.ToString(), 186 infoDesc = "反馈报文", 187 serial = Guid.NewGuid().ToString("N") 188 }; 189 ackMessage.content.oriSerial = oriSerial; 190 var sendStr = $"?{JsonConvert.SerializeObject(ackMessage)}$"; 191 192 var infoDisplay = socketTool.CreateInfoDisplay(socket); 193 infoDisplay.Message = sendStr; 194 infoDisplay.CustomColor = Color.DodgerBlue; 195 socketTool.PrintInfoConsole($"{sendStr}", ConsoleColor.Green, infoDisplay, UpdateUi.PostMessageInfo); 196 return sendStr; 197 } 198 private void ProcessReceive(SocketAsyncEventArgs e) 199 { 200 if (e.BytesTransferred > 0) 201 { 202 if (e.SocketError == SocketError.Success) 203 { 204 var data = Encoding.UTF8.GetString(e.Buffer, e.Offset, e.BytesTransferred); 205 _buffers.Append(data); 206 207 if (e.AcceptSocket.Available == 0) 208 { 209 var receStr=_buffers.ToString(); 210 if (InstanceSocketTool.ValidateMessageJson(ref receStr) == false) 211 return; 212 var info = InstanceSocketTool.CreateInfoDisplay(e.AcceptSocket); 213 info.Message = receStr; 214 info.CustomColor = Color.Green; 215 InstanceSocketTool.PrintInfoConsole($"远处：{e.AcceptSocket} 发来消息:{receStr}", Console.ForegroundColor, 216 info, UpdateUi.PostMessageInfo); 217 var receiveContent = InstanceSocketTool.UnPackMessage(receStr); 218 dynamic obj = JsonConvert.DeserializeObject(receiveContent); 219 InstanceSocketTool.ValidateMessageObj(obj, receStr); 220 AddDictWcs((int)obj.clientID.Value, e.AcceptSocket); 221 if (obj.infoType.Value != 0) 222 { 223 ReceiveMailBox.Enqueue(new ReceiveEntity 224 { 225 Client = e.AcceptSocket, 226 Message = obj 227 }); 228 } 229 //重置 230 _buffers = new StringBuilder(); 231 //发送反馈 232 var sendStr = ReplyAckWcs(e.AcceptSocket, Convert.ToString(obj.serial.Value)); 233 var sendBuffer = (byte[])Encoding.UTF8.GetBytes(sendStr); 234 e.SetBuffer(sendBuffer, 0, sendBuffer.Length); 235 if (!e.AcceptSocket.SendAsync(e)) 236 { 237 SendRequest(obj, e.AcceptSocket, e); 238 ProcessSend(e); 239 } 240 } 241 else if (!e.AcceptSocket.ReceiveAsync(e)) 242 { 243 ProcessReceive(e); 244 } 245 } 246 else 247 { 248 //this.ProcessError(e); 249 AppLogger.Error(e.SocketError.ToString()); 250 } 251 } 252 else 253 { 254 //this.CloseClientSocket(e); 255 } 256 257 } 258 static void Callback(IAsyncResult result) 259 { 260 261 262 } 263 264 private void SendRequest(dynamic Message,Socket Client, SocketAsyncEventArgs e) 265 { 266 var wcsReceiver = new WcsReceiver(TaskStockIn) 267 { 268 CacheSyncPool = SPoolyncDataPool, 269 ClientId = Convert.ToString(Message.serial.Value) 270 }; 271 272 if (Message.infoType == 40 || Message.infoType == 42) 273 { 274 wcsReceiver.ReplyBrowser(Message); 275 return; 276 } 277 wcsReceiver.RecieiveRequest(Message, new Action(delegate 278 { 279 280 281 })); 282 283 var packageList = new FrameHandlerTool().GetPackage(Message); 284 var sleepTime = 2000; 285 if (packageList.Count == 1) 286 sleepTime = 1; 287 foreach (var message in packageList) 288 { 289 Thread.Sleep(sleepTime); 290 var sendStr = $"?{JsonConvert.SerializeObject(message)}$"; 291 var sendBuffer = (byte[])Encoding.UTF8.GetBytes(sendStr); 292 e.SetBuffer(sendBuffer, 0, sendBuffer.Length); 293 while (e.AcceptSocket.SendAsync(e)) 294 { 295 } 296 var infoDisplay = InstanceSocketTool.CreateInfoDisplay(e.AcceptSocket); 297 infoDisplay.Message = sendStr; 298 infoDisplay.CustomColor = Color.DodgerBlue; 299 InstanceSocketTool.PrintInfoConsole($"{sendStr}", ConsoleColor.Green, infoDisplay, UpdateUi.PostMessageInfo); 300 } 301 } 302 public void HandReceiveMailbox() 303 { 304 while (true) 305 { 306 if (ReceiveMailBox.IsEmpty) 307 { 308 Thread.Sleep(100); 309 continue; 310 } 311 try 312 { 313 ReceiveEntity messageEntity; 314 ReceiveMailBox.TryDequeue(out messageEntity); 315 316 Action a = () => 317 { 318 var wcsReceiver = new WcsReceiver(TaskStockIn) 319 { 320 CacheSyncPool = SPoolyncDataPool, 321 ClientId = Convert.ToString(messageEntity.Message.serial.Value) 322 }; 323 324 if (messageEntity.Message.infoType == 40 || messageEntity.Message.infoType == 42) 325 { 326 wcsReceiver.ReplyBrowser(messageEntity.Message); 327 return; 328 } 329 wcsReceiver.RecieiveRequest(messageEntity.Message, new Action(delegate 330 { 331 })); 332 SendMailBox.Enqueue(new ReceiveEntity 333 { 334 Client = messageEntity.Client, 335 Message = wcsReceiver.Message 336 }); 337 338 }; 339 a.BeginInvoke(new AsyncCallback(Callback), null); 340 if (ReceiveMailBox.IsEmpty) 341 { 342 ReceiveMailBox = new ConcurrentQueue<ReceiveEntity>(); 343 ClearMemory(); 344 } 345 } 346 catch (Exception ex) 347 { 348 AppLogger.Error($"{ex.Message} {ex.StackTrace}", ex); 349 } 350 } 351 } 352 353 public void HandleSendMailbox() 354 { 355 while (true) 356 { 357 try 358 { 359 if (SendMailBox.IsEmpty) 360 { 361 Thread.Sleep(100); 362 continue; 363 } 364 ReceiveEntity messageEntity; 365 SendMailBox.TryDequeue(out messageEntity); 366 var wcsReceiver = new WcsReceiver(TaskStockIn) 367 { 368 Message = messageEntity.Message 369 }; 370 wcsReceiver.ReplyResponseToWcs(messageEntity.Client); 371 if (SendMailBox.IsEmpty) 372 { 373 ClearMemory(); 374 } 375 } 376 catch (Exception ex) 377 { 378 AppLogger.Error($"HandleSendMailbox：{ex.StackTrace}【-】{ex.Message}", ex); 379 } 380 } 381 } 382 #region 内存回收 383 [DllImport("kernel32.dll", EntryPoint = "SetProcessWorkingSetSize")] 384 public static extern int SetProcessWorkingSetSize(IntPtr process, int minSize, int maxSize); 385 /// <summary> 386 /// 释放内存 387 /// </summary> 388 public static void ClearMemory() 389 { 390 GC.Collect(); 391 GC.WaitForPendingFinalizers(); 392 if (Environment.OSVersion.Platform == PlatformID.Win32NT) 393 { 394 SetProcessWorkingSetSize(Process.GetCurrentProcess().Handle, -1, -1); 395 } 396 } 397 #endregion 398 } 399 }
备注：把原来的收件与发件模型（采用同步的方式）给重构了，采用SendRequest一个函数搞定，程序测试下，慢在回发消息（即从第三方接口拉数据，需要一定时间），收到的消息，确认帧可以立发，性能貌似还可。
备注2：前期需求不明确，都没想到会有那么多代码，所以前期思考不足，程序书写没考虑那么多，直接采用同步方式，来一个客户端开启一个进程，弊端是大大的，还好业务场景相对简单，经过一番折腾，采用2个队列来解耦，程序变的好看起来了。软件嘛，就是兼具美感的艺术品，如果自己看着都不舒服，那真是到重构的时间了（哈哈哪本书好像说过）。随着抽象出报文工厂、工具处理类、消息转发路由，程序确实清晰很多，修改业务功能、重构（之前学习过模式设计以及重构，模式容易生搬硬套，要化有形于无形，重构对我帮助最大，在写代码时就考虑到重构，逼迫我写的代码不能一团浆糊，否则将来如何重构啊，能独立出函数就函数，能抽象出类则用独立类）socket处理方式时，基本不费吹灰之力，有时间就好好改代码吧。