高性能异步等待套接字

Question

我正在编写一个应用程序,需要通过tcp进行数百个套接字连接才能读/写数据.

我在这里遇到过这段代码片段,我想知道如何让它更强大.

这是我目前调用代码的方式:

foreach (var ip in listofIps)
{
   IPEndPoint remoteEP = new IPEndPoint(IPAddress.Parse(ip), 4001);
   Socket client = new Socket(AddressFamily.InterNetwork,
                           SocketType.Stream, ProtocolType.Tcp);
   client.Connect(remoteEP);
   await ReadAsync(client);
}

1. 以上是否有任何问题,如何进行优化以使其同时运行？

   在代码片段中,缓冲区大小设置为1000.正如一个简单的说明,如果我只尝试打印出接收的字节,而不是剩余的0x00,我必须做这样的事情:

   while (true)
   {
       await s.ReceiveAsync(awaitable);
       int bytesRead = args.BytesTransferred;
       if (bytesRead <= 0) break;
       var hex = new StringBuilder(bytesRead * 2);
       var msg = new byte[bytesRead];
   
       for (int i = 0; i < bytesRead; i++)                
           msg[i] = args.Buffer[i];                
   
       foreach (byte b in msg)                
           hex.AppendFormat("{0:x2} ", b);
   
       AppendLog(string.Format("RX: {0}", hex));
   }
   

2. 有更有效的方法吗？以前,我会迭代整个缓冲区并打印出数据,但这会给我一大堆尾随的0x00,因为我的协议长度在60到70个字节之间.

Answer 1

我正在编写一个应用程序,需要通过tcp进行数百个套接字连接才能读/写数据.

你不需要"高性能套接字".使用常规性能套接字,代码简单得多.

对于初学者,请不要使用您发布的链接中的自定义等待项.它们对某些人来说非常好(并且完全"强大"),但是你不需要它们,没有它们你的代码会更简单.

1. 以上是否有任何问题,是否可以进一步优化？

是.您不应该混合使用blocking(Connect)和asynchronous(ReadAsync)代码.我会推荐这样的东西:

foreach (var ip in listofIps)
{
  IPEndPoint remoteEP = new IPEndPoint(IPAddress.Parse(ip), 4001);
  Socket client = new Socket(AddressFamily.InterNetwork,
                             SocketType.Stream, ProtocolType.Tcp);
  await client.ConnectTaskAsync(remoteEP);
  ...
}

哪里ConnectTaskAsync是一个标准的TAP-过APM包装:

public static Task ConnectTaskAsync(this Socket socket, EndPoint endpoint)
{
  return TaskFactory.FromAsync(socket.BeginConnect, socket.EndConnect, endpoint, null);
}

正如Marc Gravell指出的那样,这段代码(和您的原始代码)一次连接一个套接字.您可以使用Task.WhenAll它们同时连接它们.

2)有更有效的方法吗？

首先,您应该定义ReceiveTaskAsync类似于上面的TAP-over-APM 包装器.在处理二进制数据时,我还想在字节数组上使用扩展方法进行转储:

public string DumpHex(this ArraySegment<byte> data)
{
  return string.Join(" ", data.Select(b => b.ToString("X2")));
}

然后你可以得到这样的代码:

while (true)
{
  int bytesRead = await socket.ReceiveTaskAsync(buffer);
  if (bytesRead == 0) break;
  var data = new ArraySegment<byte>(buffer, 0, bytesRead);
  AppendLog("RX: " + data.HexDump());
  ...
}

如果你做了很多二进制操作,你会发现我的ArraySegments库很有帮助.

3)我应该在哪里以及如何包含逻辑来检查我的整个数据是否已经在一次读取中到达

哦,它比那更复杂.:)套接字是流抽象,而不是消息抽象.因此,如果要在协议中定义"消息",则需要包含长度前缀或分隔符字节,以便可以检测消息边界.然后你需要编写将解析你的消息的代码,记住从套接字读取的数据块可能只包含部分消息(所以你必须缓冲它),一个完整的消息,多个完整的消息,也可能以部分消息结束(再次,缓冲).在接收新块时,您还必须考虑现有缓冲区.

我有一个TCP/IP套接字.NET常见问题在我的博客是专门针对此,有一些示例代码使用消息框(4个字节的小端长度前缀)我个人的缺省优先级.

4)我应该如何包含writeasync方法,以便我可以在读取过程中通过套接字发送数据.

那个令人惊讶的棘手:

public static Task<int> SendTaskAsync(this Socket socket, byte[] buffer, int offset, int size, SocketFlags flags)
{
  return Task<int>.Factory.FromAsync(socket.BeginSend, socket.EndSend, buffer, offset, size, flags, null);
}
public static Task WriteAsync(this Socket socket, byte[] buffer)
{
  int bytesSent = 0;
  while (bytesSent != buffer.Length)
  {
    bytesSent += await socket.SendTaskAsync(data, bytesSent, buffer.Length - bytesSent, SocketFlags.None);
  }
}