最近在为公司的分布式服务框架做支持异步调用的开发,这种新特性的上线需要进行各种严格的测试。在并发性能测试时,性能一直非常差,而且非常的不稳定。经过不断的分析调优,发现Socket通信和多线程异步回调存在较为严重的性能问题。经过多方优化,性能终于达标。下面是原版本、支持异步最初版本和优化后版本的性能比较。差异还是非常巨大的。另外说明一下,总耗时是指10000次请求累计执行时间。
从上图可以看到,支持异步的版本,在单线程模式下,性能的表现与老版本差异并不明显,但是10线程下差异就非常巨大,而100线程的测试结果反而有所好转。通过分析,两个版本的性能差异如此巨大,主要是因为:
-
同步模式会阻塞客户端请求,说白了,在线程内就是串行请求的。但是在异步模式中,线程内的请求不再阻塞,网络流量、后台计算压力瞬间暴涨,峰值是同步模式的100倍。网络传输变成瓶颈点。
-
在压力暴涨的情况下,CPU资源占用也会突变, 并且ThreadPool、Task、异步调用的执行都将变慢。
在网络通信方面,把原先半异步的模式调整为了SocketAsyncEventArgs 模式。下面是Socket通信的几种模型的介绍和示例,总结一下,与大家分享。下次再与大家分享,并发下异步调用的性能优化方案。
APM方式: Asynchronous Programming Model
异步编程模型是一种模式,该模式允许用更少的线程去做更多的操作,.NET Framework很多类也实现了该模式,同时我们也可以自定义类来实现该模式。NET Framework中的APM也称为Begin/End模式。此种模式下,调用BeginXXX方法来启动异步操作,然后返回一个IAsyncResult 对象。当操作执行完成后,系统会触发IAsyncResult 对象的执行。 具体可参考: https://docs.microsoft.com/en-us/dotnet/standard/asynchronous-programming-patterns/asynchronous-programming-model-apm
.net中的Socket异步模式也支持APM,与同步模式或Blocking模式相比,可以更好的利用网络带宽和系统资源编写出具有更高性能的程序。参考具体代码如下:
服务端监听:
Socket serverSocket =
new
Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
IPEndPoint serverIP =
new
IPEndPoint(IPAddress.Any, 9050);
serverSocket.Bind(serverIP);
serverSocket.BeginAccept(ar =>
base
.communicateSocket = serverSocket.EndAccept(ar);
客户端连接:
var communicateSocket =
new
Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
communicateSocket.Bind(
new
IPEndPoint(IPAddress.Any, 9051));
//服务器的IP和端口
IPEndPoint serverIP;
try
{
serverIP =
new
IPEndPoint(IPAddress.Parse(IP), 9050);
}
catch
{
throw
new
Exception(String.Format(
"{0}不是一个有效的IP地址!"
, IP));
}
//客户端只用来向指定的服务器发送信息,不需要绑定本机的IP和端口,不需要监听
try
c
ommunicateSocket.BeginConnect(serverIP, ar =>
{
AccessAciton();
},
null
);
}
catch
{
throw
new
Exception(
string
.Format(
"尝试连接{0}不成功!"
, IP));
}
客户端请求:
if
(communicateSocket.Connected ==
false
)
{
throw
new
Exception(
"还没有建立连接, 不能发送消息"
);
}
Byte[] msg = Encoding.UTF8.GetBytes(message);
communicateSocket.BeginSend(msg,0, msg.Length, SocketFlags.None,
ar => {
},
null
);
服务端响应:
Byte[] msg =
new
byte
[1024];
//异步的接受消息
communicateSocket.BeginReceive(msg, 0, msg.Length, SocketFlags.None,
ar => {
//对方断开连接时, 这里抛出Socket Exception
communicateSocket.EndReceive(ar);
ReceiveAction(Encoding.UTF8.GetString(msg).Trim(
'\0'
,
' '
));
Receive(ReceiveAction);
注意:异步模式虽好,但是如果进行大量异步套接字操作,是要付出很高代价的。针对每次操作,都必须创建一个IAsyncResult对象,而且该对象不能被重复使用。由于大量使用对象分配和垃圾收集,这会影响系统性能。如需要更好的理解APM模式,最了解EAP模式:Event-based Asynchronous Pattern:https://docs.microsoft.com/en-us/dotnet/standard/asynchronous-programming-patterns/event-based-asynchronous-pattern-eap 。
TAP 方式: Task-based Asynchronous Pattern
基于任务的异步模式,该模式主要使用System.Threading.Tasks.Task和Task<T>类来完成异步编程,相对于APM 模式来讲,TAP使异步编程模式更加简单(因为这里我们只需要关注Task这个类的使用),同时TAP也是微软推荐使用的异步编程模式。APM与TAP的本质区别,请参考我的一篇历史博客:http://www.cnblogs.com/vveiliang/p/7943003.html
TAP模式与APM模式是两种异步模式的实现,从性能上看没有本质的差别。TAP的资料可参考:https://docs.microsoft.com/en-us/dotnet/standard/asynchronous-programming-patterns/task-based-asynchronous-pattern-tap 。参考具体代码如下:
服务端:
publicclassStateContext
{
// Client socket.
publicSocketWorkSocket =null;
// Size of receive buffer.
publicconstintBufferSize = 1024;
// Receive buffer.
publicbyte[] buffer =newbyte[BufferSize];
// Received data string.
publicStringBuildersb =newStringBuilder(100);
}
publicclassAsynchronousSocketListener
{
// Thread signal.
publicstaticManualResetEventreSetEvent =newManualResetEvent(false);
publicAsynchronousSocketListener()
{
}
publicstaticvoidStartListening()
{
// Data buffer for incoming data.
byte[] bytes =newByte[1024];
// Establish the local endpoint for the socket.
IPAddressipAddress =IPAddress.Parse("127.0.0.1");
IPEndPointlocalEndPoint =newIPEndPoint(ipAddress, 11000);
// Create a TCP/IP socket.
Socketlistener =newSocket(AddressFamily.InterNetwork,SocketType.Stream,ProtocolType.Tcp);
// Bind the socket to the local
try
{
listener.Bind(localEndPoint);
listener.Listen(100);
while(true)
{
// Set the event to nonsignaled state.
reSetEvent.Reset();
// Start an asynchronous socket to listen for connections.
Console.WriteLine("Waiting for a connection...");
listener.BeginAccept(newAsyncCallback(AcceptCallback), listener);
// Wait until a connection is made before continuing.
reSetEvent.WaitOne();
}
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
Console.WriteLine("\nPress ENTER to continue...");
Console.Read();
}
publicstaticvoidAcceptCallback(IAsyncResultar)
{
// Signal the main thread to continue.
reSetEvent.Set();
// Get the socket that handles the client request.
Socketlistener = (Socket)ar.AsyncState;
Sockethandler = listener.EndAccept(ar);
// Create the state object.
StateContextstate =newStateContext();
state.WorkSocket = handler;
handler.BeginReceive(state.buffer, 0,StateContext.BufferSize, 0,newAsyncCallback(ReadCallback), state);
}
publicstaticvoidReadCallback(IAsyncResultar)
{
Stringcontent =String.Empty;
StateContextstate = (StateContext)ar.AsyncState;
Sockethandler = state.WorkSocket;
// Read data from the client socket.
intbytesRead = handler.EndReceive(ar);
if(bytesRead > 0)
{
// There might be more data, so store the data received so far.
state.sb.Append(Encoding.ASCII.GetString(state.buffer, 0, bytesRead));
// Check for end-of-file tag. If it is not there, read
// more data.
content = state.sb.ToString();
if(content.IndexOf("<EOF>") > -1)
{
Console.WriteLine("读取 {0} bytes. \n 数据: {1}", content.Length, content);
Send(handler, content);
}
else
{
handler.BeginReceive(state.buffer, 0,StateContext.BufferSize, 0,newAsyncCallback(ReadCallback), state);
}
}
}
privatestaticvoidSend(Sockethandler,Stringdata)
{
byte[] byteData =Encoding.ASCII.GetBytes(data);
handler.BeginSend(byteData, 0, byteData.Length, 0,newAsyncCallback(SendCallback), handler);
}
privatestaticvoidSendCallback(IAsyncResultar)
{
try
{
Sockethandler = (Socket)ar.AsyncState;
intbytesSent = handler.EndSend(ar);
Console.WriteLine("发送 {0} bytes.", bytesSent);
handler.Shutdown(SocketShutdown.Both);
handler.Close();
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
}
publicstaticintMain(String[] args)
{
StartListening();
return0;
客户端:
publicclassAsynchronousClient
{
// The port number for the remote device.
privateconstintport = 11000;
// ManualResetEvent instances signal completion.
privatestaticManualResetEventconnectResetEvent =newManualResetEvent(false);
privatestaticManualResetEventsendResetEvent =newManualResetEvent(false);
privatestaticManualResetEventreceiveResetEvent =newManualResetEvent(false);
privatestaticStringresponse =String.Empty;
privatestaticvoidStartClient()
{
try
{
IPAddressipAddress =IPAddress.Parse("127.0.0.1");
IPEndPointremoteEP =newIPEndPoint(ipAddress, port);
// Create a TCP/IP socket.
Socketclient =newSocket(AddressFamily.InterNetwork,SocketType.Stream,ProtocolType.Tcp);
// Connect to the remote endpoint.
client.BeginConnect(remoteEP,newAsyncCallback(ConnectCallback), client);
connectResetEvent.WaitOne();
Send(client,"This is a test<EOF>");
sendResetEvent.WaitOne();
Receive(client);
receiveResetEvent.WaitOne();
Console.WriteLine("Response received : {0}", response);
// Release the socket.
client.Shutdown(SocketShutdown.Both);
client.Close();
Console.ReadLine();
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
}
privatestaticvoidConnectCallback(IAsyncResultar)
{
try
{
Socketclient = (Socket)ar.AsyncState;
client.EndConnect(ar);
Console.WriteLine("Socket connected to {0}", client.RemoteEndPoint.ToString());
connectResetEvent.Set();
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
}
privatestaticvoidReceive(Socketclient)
{
try
{
StateContextstate =newStateContext();
state.WorkSocket = client;
client.BeginReceive(state.buffer, 0,StateContext.BufferSize, 0,newAsyncCallback(ReceiveCallback), state);
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
}
privatestaticvoidReceiveCallback(IAsyncResultar)
{
try
{
StateContextstate = (StateContext)ar.AsyncState;
Socketclient = state.WorkSocket;
intbytesRead = client.EndReceive(ar);
if(bytesRead > 0)
{
state.sb.Append(Encoding.ASCII.GetString(state.buffer, 0, bytesRead));
client.BeginReceive(state.buffer, 0,StateContext.BufferSize, 0,newAsyncCallback(ReceiveCallback), state);
}
else
{
if(state.sb.Length > 1)
{
response = state.sb.ToString();
}
receiveResetEvent.Set();
}
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
}
privatestaticvoidSend(Socketclient,Stringdata)
{
byte[] byteData =Encoding.ASCII.GetBytes(data);
client.BeginSend(byteData, 0, byteData.Length, 0,newAsyncCallback(SendCallback), client);
}
privatestaticvoidSendCallback(IAsyncResultar)
{
try
{
Socketclient = (Socket)ar.AsyncState;
intbytesSent = client.EndSend(ar);
Console.WriteLine("Sent {0} bytes to server.", bytesSent);
sendResetEvent.Set();
}
catch(Exceptione)
{
Console.WriteLine(e.ToString());
}
}
publicstaticintMain(String[] args)
{
StartClient();
return0;
}
SAEA方式: SocketAsyncEventArgs
APM模式、TAP模式虽然解决了Socket的并发问题,但是在大并发下还是有较大性能问题的。这主要是因为上述两种模式都会生产 IAsyncResult 等对象 ,而大量垃圾对象的回收会非常影响系统的性能。为此,微软推出了 SocketAsyncEventArgs 。SocketAsyncEventArgs 是 .NET Framework 3.5 开始支持的一种支持高性能 Socket 通信的实现。SocketAsyncEventArgs 相比于 APM 方式的主要优点可以描述如下,无需每次调用都生成 IAsyncResult 等对象,向原生 Socket 更靠近一些。这是官方的解释:
The main feature of these enhancements is the avoidance of the repeated allocation and synchronization of objects during high-volume asynchronous socket I/O. The Begin/End design pattern currently implemented by the Socket class for asynchronous socket I/O requires a System.IAsyncResult object be allocated for each asynchronous socket operation.
SocketAsyncEventArgs主要为高性能网络服务器应用程序而设计,避免了在异步套接字 I/O 量非常大时,大量垃圾对象创建与回收。使用此类执行异步套接字操作的模式包含以下步骤,具体说明可参考:https://msdn.microsoft.com/en-us/library/system.net.sockets.socketasynceventargs(v=vs.110).aspx 。
-
分配一个新的 SocketAsyncEventArgs 上下文对象,或者从应用程序池中获取一个空闲的此类对象。
-
将该上下文对象的属性设置为要执行的操作(例如,完成回调方法、数据缓冲区、缓冲区偏移量以及要传输的最大数据量)。
-
调用适当的套接字方法 (xxxAsync) 以启动异步操作。
-
如果异步套接字方法 (xxxAsync) 返回 true,则在回调中查询上下文属性来获取完成状态。
-
如果异步套接字方法 (xxxAsync) 返回 false,则说明操作是同步完成的。 可以查询上下文属性来获取操作结果。
-
将该上下文重用于另一个操作,将它放回到应用程序池中,或者将它丢弃。
下面是封装的一个组件代码:
classBufferManager
{
intm_numBytes; // the total number of bytes controlled by the buffer pool
byte[] m_buffer; // the underlying byte array maintained by the Buffer Manager
Stack<int> m_freeIndexPool; //
intm_currentIndex;
intm_bufferSize;
publicBufferManager(inttotalBytes,intbufferSize)
{
m_numBytes = totalBytes;
m_currentIndex = 0;
m_bufferSize = bufferSize;
m_freeIndexPool =newStack<int>();
}
// Allocates buffer space used by the buffer pool
publicvoidInitBuffer()
{
// create one big large buffer and divide that
// out to each SocketAsyncEventArg object
m_buffer =newbyte[m_numBytes];
}
// Assigns a buffer from the buffer pool to the
// specified SocketAsyncEventArgs object
//
// <returns>true if the buffer was successfully set, else false</returns>
publicboolSetBuffer(SocketAsyncEventArgsargs)
{
if(m_freeIndexPool.Count > 0)
{
args.SetBuffer(m_buffer, m_freeIndexPool.Pop(), m_bufferSize);
}
else
{
if((m_numBytes - m_bufferSize) < m_currentIndex)
{
returnfalse;
}
args.SetBuffer(m_buffer, m_currentIndex, m_bufferSize);
m_currentIndex += m_bufferSize;
}
returntrue;
}
// Removes the buffer from a SocketAsyncEventArg object.
// This frees the buffer back to the buffer pool
publicvoidFreeBuffer(SocketAsyncEventArgsargs)
{
m_freeIndexPool.Push(args.Offset);
args.SetBuffer(null, 0, 0);
}
}
///<summary>
///This class is used to communicate with a remote application over TCP/IP protocol.
///</summary>
classTcpCommunicationChannel
{
#regionPrivate fields
///<summary>
///Size of the buffer that is used to receive bytes from TCP socket.
///</summary>
privateconstintReceiveBufferSize = 8 * 1024;//4KB
///<summary>
///This buffer is used to receive bytes
///</summary>
privatereadonlybyte[] _buffer;
///<summary>
///Socket object to send/reveice messages.
///</summary>
privatereadonlySocket_clientSocket;
///<summary>
///A flag to control thread's running
///</summary>
privatevolatilebool_running;
///<summary>
///This object is just used for thread synchronizing (locking).
///</summary>
privatereadonlyobject_syncLock;
privateBufferManagerreceiveBufferManager;
privateSocketAsyncEventArgsreceiveBuff =null;
#endregion
#regionConstructor
///<summary>
///Creates a new TcpCommunicationChannel object.
///</summary>
///<param name="clientSocket">A connected Socket object that is
///used to communicate over network</param>
publicTcpCommunicationChannel(SocketclientSocket)
{
_clientSocket = clientSocket;
_clientSocket.Blocking =false;
_buffer =newbyte[ReceiveBufferSize];
_syncLock =newobject();
Init();
}
privatevoidInit()
{
//初始化接收Socket缓存数据
receiveBufferManager =newBufferManager(ReceiveBufferSize*2, ReceiveBufferSize);
receiveBufferManager.InitBuffer();
receiveBuff =newSocketAsyncEventArgs();
receiveBuff.Completed += ReceiveIO_Completed;
receiveBufferManager.SetBuffer(receiveBuff);
//初始化发送Socket缓存数据
}
#endregion
#regionPublic methods
///<summary>
///Disconnects from remote application and closes channel.
///</summary>
publicvoidDisconnect()
{
_running =false;
receiveBuff.Completed -= ReceiveIO_Completed;
receiveBuff.Dispose();
if(_clientSocket.Connected)
{
_clientSocket.Close();
}
_clientSocket.Dispose();
}
#endregion
publicvoidStartReceive()
{
_running =true;
boolresult = _clientSocket.ReceiveAsync(receiveBuff);
}
privatevoidReceiveIO_Completed(objectsender,SocketAsyncEventArgse)
{
if(e.BytesTransferred > 0 && e.SocketError ==SocketError.Success && _clientSocket.Connected ==true&& e.LastOperation ==SocketAsyncOperation.Receive)
{
if(!_running)
{
return;
}
//Get received bytes count
DateTimereceiveTime =DateTime.Now;
//Copy received bytes to a new byte array
varreceivedBytes =newbyte[e.BytesTransferred];
Array.Copy(e.Buffer, 0, receivedBytes, 0, e.BytesTransferred);
//处理消息....
if(_running)
{
StartReceive();
}
}
}
///<summary>
///Sends a message to the remote application.
///</summary>
///<param name="message">Message to be sent</param>
publicvoidSendMessage(byte[] messageBytes)
{
//Send message
if(_clientSocket.Connected)
{
SocketAsyncEventArgsdata =newSocketAsyncEventArgs();
data.SocketFlags =SocketFlags.None;
data.Completed += (s, e) =>
{
e.Dispose();
};
data.SetBuffer(messageBytes, 0, messageBytes.Length);
//Console.WriteLine("发送:" + messageBytes.LongLength);
_clientSocket.SendAsync(data);
}
}