基于SocketAsyncEventArgs的版本

时间:2023-03-09 01:19:22
基于SocketAsyncEventArgs的版本

分享一下,项目中写的简单socket程序,不同方式的版本,这是个异步基于IOCP实现高性能的处理方式。IOCP就不多说了,高性能的完成端口,可以实现套接字对象的复用,降低开销,且基于端口共享性能据说很高,因此把之前的项目做个改动。下面来个代码走读:

功能介绍:

此socket程序类似一个交换机,对下统一采用报文来交换数据,什么心跳、回执等各式各样十几种类,他负责接收各个client发送来的消息,向各个client回发确认帧、发送对设备任务请求等工作;对上,向数据库存入设备状态等数据,将应用层的任务,组织合适的报文,下发给client。

他所处的角色和位置:有时候,接收到client的请求后直接转发给BS(本质就是存入数据库啦);有时候,接收到client的请求后 思考 一下,向应用层拉取数据,装配为 恰当的报文 后,回送给对应的client;轮询BS提供的业务接口,发现有对设备请求的任务后,装配为恰当的格式后,发送给对应的client。

就像一个繁忙的机械手,有时对于收到的报文,仅仅是左手接,逻辑处理后,右手丢回去,必须满足,从哪里接的、需送回给正确的client;有时候,只是充当一个转发器,收到的内容不做任何逻辑处理,直接丢给第三方处理。虽然程序都已经运行多时,都没有想好一个恰当的名字来命名程序,他更像是一个桥、一个转换器。

下面是主进程需要声明的符号;收件箱存入接收的消息,不做任何处理;会有相应的进程不间断的扫描收件队列,一旦发现内容,便开始处理,处理结果会存入对应的发件箱,同样有相应的进程不间断的处理,通常为发送至client端,或者发送至BS接口。

  private readonly Config _config = InitConfig.ReadConfig();//配置信息 端口 IP

         private static IDictionary<int, WcsEndpoint<Socket>> _wcsList = new Dictionary<int, WcsEndpoint<Socket>>();//记录有多少个端点连接进来
private const int BufferSize = ;
private static SocketTool _socketTool = new SocketTool();//完成分包、压缩、加密关键报文内容
private static MessageFactory _messageFactory = new MessageFactory();//用于构造不同的消息报文
private static TaskSendDown _taskStockIn = new TaskSendDown(_socketTool);//用于桥接浏览器,通知web端一些事情发生
private static ConcurrentQueue<ReceiveEntity> _receiveMailBox = new ConcurrentQueue<ReceiveEntity>();//收件箱
private static ConcurrentQueue<ReceiveEntity> _sendMailBox = new ConcurrentQueue<ReceiveEntity>();//发件箱
private static object _lockwcsList = new object();//同步锁
private static CachePool _sPoolyncDataPool = new CachePool();//扩展数据缓存
  var ip = IPAddress.Parse(_config.LocalIp);
TcpListener.InstanceSocketTool = _socketTool;
TcpListener.LockwcsList = _lockwcsList;
TcpListener.WcsList = _wcsList;
var core = new TcpListener
{//将外部变量传递给核心处理类
MaxConnect = int.Parse(_config.MaxConnect),
ReceiveMailBox = _receiveMailBox,
SPoolyncDataPool = _sPoolyncDataPool,
SendMailBox = _sendMailBox,
TaskStockIn = _taskStockIn,
};
CallbackUpdateStrip("启动wcs监听;");//同步UI
var receiveBoxHandler = new Thread(core.HandReceiveMailbox); //收件邮箱侦听
receiveBoxHandler.Start(); var sendBoxHandler = new Thread(core.HandleSendMailbox); //发件箱侦听
sendBoxHandler.Start();
core.Listen(new IPEndPoint(ip, int.Parse(_config.PortForWcs)));

核心socket实现,抄袭了别人的稍微改造一下,或者高端点重构了一下:

 using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Drawing;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Runtime.Remoting.Contexts;
using System.Text;
using System.Threading;
using Newtonsoft.Json;
using NovaMessageSwitch;
using NovaMessageSwitch.Bll;
using NovaMessageSwitch.message;
using NovaMessageSwitch.Model;
using NovaMessageSwitch.Tool;
using NovaMessageSwitch.Tool.DataCache;
using NovaMessageSwitch.Tool.Log; /****************************************************************
* 作者:wxy
* CLR版本:4.0.30319.42000
* 创建时间:2016/3/25 8:47:51
* 2016
* 描述说明:
* SocketAsyncEventArgs是异步的socket类,封装了IOCP,可以很方便实现non-blocking IO
非阻塞IO对server性能和吞吐量有很大好处
* 修改历史:
* IPAddress[] addressList = Dns.GetHostEntry(Environment.MachineName).AddressList;
new TcpListener().Listen(/new IPEndPoint(addressList[addressList.Length - 1]/, 9900)); Console.ReadKey(); *
*****************************************************************/
namespace CompletePortB
{
public class TcpListener
{
private SocketAsyncEventArgs _args;
private Socket _listenerSocket;
private StringBuilder _buffers;
private const int BufferSize = ; public TcpListener()
{
MaxConnect = ; } public int MaxConnect { get; set; }
public static object LockwcsList { get; set; }
public static IDictionary<int, WcsEndpoint<Socket>> WcsList { get; set; }
public TaskSendDown TaskStockIn { get; set; }
public CachePool SPoolyncDataPool { get; set; }
public ConcurrentQueue<ReceiveEntity> ReceiveMailBox { get; set; }
public ConcurrentQueue<ReceiveEntity> SendMailBox { get; set; } public static SocketTool InstanceSocketTool { get; set; } public void Listen(EndPoint e)
{
try
{
_buffers = new StringBuilder();
_listenerSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_listenerSocket.Bind(e);
_listenerSocket.Listen(MaxConnect);
_args = new SocketAsyncEventArgs();
_args.Completed += new EventHandler<SocketAsyncEventArgs>(ProcessAccept);
BeginAccept(_args);
Console.Read();
}
catch (Exception ex)
{ }
} private void BeginAccept(SocketAsyncEventArgs e)
{
e.AcceptSocket = null;
//异步操作完成,返回false
if (!_listenerSocket.AcceptAsync(e))
ProcessAccept(_listenerSocket, e); } //异步操作完成时调用此方法
private void ProcessAccept(object sender, SocketAsyncEventArgs e)
{
var s = e.AcceptSocket;
e.AcceptSocket = null;
var args = new SocketAsyncEventArgs();
//继续向下回掉函数
args.Completed += new EventHandler<SocketAsyncEventArgs>(OnIoCompleted);
//设定异步套接字方法的数据缓冲区
args.SetBuffer(new byte[BufferSize], , BufferSize);
args.AcceptSocket = s;
//IO操作同步完成,返回false,且不引发e参数的socketasynceventargs事件
//IO操作若被挂起,将返回true,且引发e参数的...
if (!s.ReceiveAsync(args))
ProcessReceive(args);
BeginAccept(e);
}
void OnIoCompleted(object sender, SocketAsyncEventArgs e)
{
//获取最近使用此上下文对象执行的套接字操作类型
switch (e.LastOperation)
{
case SocketAsyncOperation.Receive:
this.ProcessReceive(e);
break;
case SocketAsyncOperation.Send:
this.ProcessSend(e);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a receive or send");
}
} private void ProcessSend(SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.Success)
{
//接收完毕开始下次接收
if (!e.AcceptSocket.ReceiveAsync(e))
{
this.ProcessReceive(e);
}
}
else
{ } } /// <summary>
/// 维护wcs端点列表
/// </summary>
/// <param name="clientId"></param>
/// <param name="socket"></param>
private static void AddDictWcs(int clientId, Socket socket)
{
lock (LockwcsList)
{
if (WcsList.Keys.Contains(clientId))
{
var endPoint = WcsList[clientId];
endPoint.EndPoint = socket;
endPoint.RecentTimeOld = endPoint.RecentTime;
endPoint.RecentTime = DateTime.Now;
InstanceSocketTool.UpdateWcsDisplay(endPoint, UpdateUi.Post);
}
else
{
var newEndPoit = new WcsEndpoint<Socket>
{
RecentTime = DateTime.Now,
RecentTimeOld = null,
EndPoint = socket
};
WcsList.Add(clientId, newEndPoit);
} foreach (var wcs in WcsList.Where(wcs => (wcs.Value.RecentTime - DateTime.Now).Minutes > ))
{
WcsList.Remove(wcs.Key);
}
InstanceSocketTool.UpdateWcsDisplay(WcsList, UpdateUi.Post);
}
}
private string ReplyAckWcs(Socket socket, string oriSerial)
{
var socketTool = InstanceSocketTool;
var ackMessage = new MessageData<ContentReply>
{
infoType = ,
content = new ContentReply(),
destination = DataFlowDirection.wcs.ToString(),
source = DataFlowDirection.wms.ToString(),
infoDesc = "反馈报文",
serial = Guid.NewGuid().ToString("N")
};
ackMessage.content.oriSerial = oriSerial;
var sendStr = $"?{JsonConvert.SerializeObject(ackMessage)}$"; var infoDisplay = socketTool.CreateInfoDisplay(socket);
infoDisplay.Message = sendStr;
infoDisplay.CustomColor = Color.DodgerBlue;
socketTool.PrintInfoConsole($"{sendStr}", ConsoleColor.Green, infoDisplay, UpdateUi.PostMessageInfo);
return sendStr;
}
private void ProcessReceive(SocketAsyncEventArgs e)
{
if (e.BytesTransferred > )
{
if (e.SocketError == SocketError.Success)
{
var data = Encoding.UTF8.GetString(e.Buffer, e.Offset, e.BytesTransferred);
_buffers.Append(data); if (e.AcceptSocket.Available == )
{
var receStr=_buffers.ToString();
if (InstanceSocketTool.ValidateMessageJson(ref receStr) == false)
return;
var info = InstanceSocketTool.CreateInfoDisplay(e.AcceptSocket);
info.Message = receStr;
info.CustomColor = Color.Green;
InstanceSocketTool.PrintInfoConsole($"远处:{e.AcceptSocket} 发来消息:{receStr}", Console.ForegroundColor,
info, UpdateUi.PostMessageInfo);
var receiveContent = InstanceSocketTool.UnPackMessage(receStr);
dynamic obj = JsonConvert.DeserializeObject(receiveContent);
InstanceSocketTool.ValidateMessageObj(obj, receStr);
AddDictWcs((int)obj.clientID.Value, e.AcceptSocket);
if (obj.infoType.Value != )
{
ReceiveMailBox.Enqueue(new ReceiveEntity
{
Client = e.AcceptSocket,
Message = obj
});
}
//重置
_buffers = new StringBuilder();
//发送反馈
var sendStr = ReplyAckWcs(e.AcceptSocket, Convert.ToString(obj.serial.Value));
var sendBuffer = (byte[])Encoding.UTF8.GetBytes(sendStr);
e.SetBuffer(sendBuffer, , sendBuffer.Length);
if (!e.AcceptSocket.SendAsync(e))
{
SendRequest(obj, e.AcceptSocket, e);
ProcessSend(e);
}
}
else if (!e.AcceptSocket.ReceiveAsync(e))
{
ProcessReceive(e);
}
}
else
{
//this.ProcessError(e);
AppLogger.Error(e.SocketError.ToString());
}
}
else
{
//this.CloseClientSocket(e);
} }
static void Callback(IAsyncResult result)
{ } private void SendRequest(dynamic Message,Socket Client, SocketAsyncEventArgs e)
{
var wcsReceiver = new WcsReceiver(TaskStockIn)
{
CacheSyncPool = SPoolyncDataPool,
ClientId = Convert.ToString(Message.serial.Value)
}; if (Message.infoType == || Message.infoType == )
{
wcsReceiver.ReplyBrowser(Message);
return;
}
wcsReceiver.RecieiveRequest(Message, new Action(delegate
{ })); var packageList = new FrameHandlerTool().GetPackage(Message);
var sleepTime = ;
if (packageList.Count == )
sleepTime = ;
foreach (var message in packageList)
{
Thread.Sleep(sleepTime);
var sendStr = $"?{JsonConvert.SerializeObject(message)}$";
var sendBuffer = (byte[])Encoding.UTF8.GetBytes(sendStr);
e.SetBuffer(sendBuffer, , sendBuffer.Length);
while (e.AcceptSocket.SendAsync(e))
{
}
var infoDisplay = InstanceSocketTool.CreateInfoDisplay(e.AcceptSocket);
infoDisplay.Message = sendStr;
infoDisplay.CustomColor = Color.DodgerBlue;
InstanceSocketTool.PrintInfoConsole($"{sendStr}", ConsoleColor.Green, infoDisplay, UpdateUi.PostMessageInfo);
}
}
public void HandReceiveMailbox()
{
while (true)
{
if (ReceiveMailBox.IsEmpty)
{
Thread.Sleep();
continue;
}
try
{
ReceiveEntity messageEntity;
ReceiveMailBox.TryDequeue(out messageEntity); Action a = () =>
{
var wcsReceiver = new WcsReceiver(TaskStockIn)
{
CacheSyncPool = SPoolyncDataPool,
ClientId = Convert.ToString(messageEntity.Message.serial.Value)
}; if (messageEntity.Message.infoType == || messageEntity.Message.infoType == )
{
wcsReceiver.ReplyBrowser(messageEntity.Message);
return;
}
wcsReceiver.RecieiveRequest(messageEntity.Message, new Action(delegate
{
}));
SendMailBox.Enqueue(new ReceiveEntity
{
Client = messageEntity.Client,
Message = wcsReceiver.Message
}); };
a.BeginInvoke(new AsyncCallback(Callback), null);
if (ReceiveMailBox.IsEmpty)
{
ReceiveMailBox = new ConcurrentQueue<ReceiveEntity>();
ClearMemory();
}
}
catch (Exception ex)
{
AppLogger.Error($"{ex.Message} {ex.StackTrace}", ex);
}
}
} public void HandleSendMailbox()
{
while (true)
{
try
{
if (SendMailBox.IsEmpty)
{
Thread.Sleep();
continue;
}
ReceiveEntity messageEntity;
SendMailBox.TryDequeue(out messageEntity);
var wcsReceiver = new WcsReceiver(TaskStockIn)
{
Message = messageEntity.Message
};
wcsReceiver.ReplyResponseToWcs(messageEntity.Client);
if (SendMailBox.IsEmpty)
{
ClearMemory();
}
}
catch (Exception ex)
{
AppLogger.Error($"HandleSendMailbox:{ex.StackTrace}【-】{ex.Message}", ex);
}
}
}
#region 内存回收
[DllImport("kernel32.dll", EntryPoint = "SetProcessWorkingSetSize")]
public static extern int SetProcessWorkingSetSize(IntPtr process, int minSize, int maxSize);
/// <summary>
/// 释放内存
/// </summary>
public static void ClearMemory()
{
GC.Collect();
GC.WaitForPendingFinalizers();
if (Environment.OSVersion.Platform == PlatformID.Win32NT)
{
SetProcessWorkingSetSize(Process.GetCurrentProcess().Handle, -, -);
}
}
#endregion
}
}

备注:把原来的收件与发件模型(采用同步的方式)给重构了,采用SendRequest一个函数搞定,程序测试下,慢在回发消息(即从第三方接口拉数据,需要一定时间),收到的消息,确认帧可以立发,性能貌似还可。

备注2:前期需求不明确,都没想到会有那么多代码,所以前期思考不足,程序书写没考虑那么多,直接采用同步方式,来一个客户端开启一个进程,弊端是大大的,还好业务场景相对简单,经过一番折腾,采用2个队列来解耦,程序变的好看起来了。软件嘛,就是兼具美感的艺术品,如果自己看着都不舒服,那真是到重构的时间了(哈哈哪本书好像说过)。随着抽象出报文工厂、工具处理类、消息转发路由,程序确实清晰很多,修改业务功能、重构(之前学习过模式设计以及重构,模式容易生搬硬套,要化有形于无形,重构对我帮助最大,在写代码时就考虑到重构,逼迫我写的代码不能一团浆糊,否则将来如何重构啊,能独立出函数就函数,能抽象出类则用独立类)socket处理方式时,基本不费吹灰之力,有时间就好好改代码吧。