用NetworkStream.ReadAsync替换Socket.ReceiveAsync(等待)

Question

我有一个应用程序,它同时生成几百个TCP连接,并从它们接收一个恒定的数据流.

 private void startReceive()
    {
        SocketAsyncEventArgs e = new SocketAsyncEventArgs();
        e.Completed += receiveCompleted;
        e.SetBuffer(new byte[1024], 0, 1024);
        if (!Socket.ReceiveAsync(e)) { receiveCompleted(this, e); }  
    }

    void receiveCompleted(object sender, SocketAsyncEventArgs e)
    {
        ProcessData(e);

        if (!Socket.ReceiveAsync(e)) { receiveCompleted(this, e); }
    }

我的尝试导致了这样的事情:

private async void StartReceive()
    {
        byte[] Buff = new byte[1024];
        int recv = 0;
        while (Socket.Connected)
        {
            recv = await NetworkStream.ReadAsync(Buff, 0, 1024);
            ProcessData(Buff,recv);
        }
    }

我遇到的问题是方法调用StartReceive()会阻塞,而不是到达随附的StartSend() method called afterStartReceive(). Creating a new task forStartReceive()would just end up with 300-ish threads, and it seems to do so just by callingStartReceive()`.

在使用a时,在我的现有代码上实现new async和await关键字的正确方法是什么？NetworkStream使用线程池Socket.SendAsync()并且Socket.ReceiveAsync()正在使用以避免必须拥有数百个线程/任务？

networkstream以这种方式使用i/o完成端口是否有任何性能优势beginreceive？

Answer 1

你在这里一次改变两件事:异步样式(SocketAsyncEventArgs到Task/ async)和抽象层次(Socketto NetworkStream).

由于您已经习惯了Socket,我建议只更改异步样式,并继续Socket直接使用该类.

Async CTP没有提供Socket任何async兼容的方法(这很奇怪;我认为它们被误删了,将被添加到.NET 4.5中).

ReceiveAsyncTask如果您使用我的AsyncEx库,那么创建自己的扩展方法(以及其他操作的类似包装器)并不难:

public static Task<int> ReceiveAsyncTask(this Socket socket,
    byte[] buffer, int offset, int size)
{
  return AsyncFactory<int>.FromApm(socket.BeginReceive, socket.EndReceive,
      buffer, offset, size, SocketFlags.None);
}

一旦你这样做,你StartReceive可以这样写:

private async Task StartReceive()
{
  try
  {
    var buffer = new byte[1024];
    while (true)
    {
      var bytesReceived = await socket.ReceiveAsyncTask(buffer, 0, 1024)
          .ConfigureAwait(false);
      ProcessData(buffer, bytesReceived);
    }
  }
  catch (Exception ex)
  {
    // Handle errors here
  }
}

现在,要解决许多小问题:

• await不会产生新的线程.我和其他许多人一样,在我的博客上写了一篇async/await简介.async/ await允许并发,但这并不一定意味着多线程.
• 数以百计的线程可能会有问题.但是,数以百计的任务根本不是问题; 线程池和BCL旨在处理许多任务.
• async/ await不是一种全新的异步处理形式; 它只是一个简单的方法来表达异步处理.它仍然使用下面的IOCP. async/的await性能略低于低级方法; 它的吸引力在于编写和编写异步方法的简易性.
• 一个非常繁忙的系统在切换到async/ 时会看到一些GC压力增加await.并行团队的Stephen Toub 写了一些示例套接字特定的等待,可以帮助解决这个问题.(我建议首先使用简单的模式,如果你觉得有必要,只使用性能增强的方法;但是,如果最终需要它,最好知道它在那里).
• 异步方法应该返回,Task除非你确实需要它们返回void.Task是等待的,所以你的方法是可组合的(更容易测试); void更像是"火与忘".
• 您可以调用ConfigureAwait(false)告诉其余async方法在线程池线程上执行.我在上面的例子中使用它,因此它ProcessData在线程池线程中执行,就像使用时一样SocketAsyncEventArgs.
• Socket.Connected没用.您需要发送数据以检测连接是否仍然有效.