对套接字的异步方法 BeginSend 的调用进行排队

Question

我需要将 BeginSend 调用排队到一个套接字，并且我需要它们按时间顺序执行。为此，我使用了一个信号量来指示回调函数何时可以执行。
大多数情况下，它之所以有效，是因为每个异步回调都在一个单独的线程上执行，但偶尔在新的异步调用中使用当前回调中使用的同一线程。发生这种情况时，该线程被锁定，等待信号量被释放，但是因为应该清除信号量的同一个线程正在等待它被清除，所以该线程被永远锁定。

为了说明问题，这里有一段测试代码：

static Semaphore semaphore = new Semaphore(1, 1);
static IList<byte[]> buffer = new List<byte[]>();
static void Main()
{
    Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
    socket.Connect(new IPEndPoint(new IPAddress(new byte[] { 192, 168, 1, 8 }), 123));
    while (true) // data feed
    {
        lock (buffer)
        {
            buffer.Add(new byte[1460]);
            if (buffer.Count == 1)
                socket.BeginSend(buffer[0], 0, 1460, 0, new AsyncCallback(SendCallback), socket); // calls BeginSend if the buffer was empty before 
        }
    }
}

static void SendCallback(IAsyncResult ar)
{
    Console.WriteLine("in " + Thread.CurrentThread.ManagedThreadId);
    semaphore.WaitOne();
    Socket socket = (Socket)ar.AsyncState;
    lock(buffer)
    {
        buffer.RemoveAt(0); // removes data that was sent
        if (buffer.Count > 0) // if there is more data to send calls BeginSend again
            socket.BeginSend(buffer[0], 0, 1460, 0, new AsyncCallback(SendCallback), socket);
    }
    semaphore.Release();
    Console.WriteLine("out " + Thread.CurrentThread.ManagedThreadId);
}

这是输出：

因为线程 10 被转移到一个新的回调，没有给前一个回调机会退出和清除信号量，线程被永远锁定。

我应该如何解决这个问题？

Answer 1

切换到任务：

不错的msdn文章Tasks and the APM Pattern

public Task<int> SendAsync(Socket socket, byte[] buffer, int offset, int size, SocketFlags flags)
{
   var result = socket.BeginSend(buffer, offset, size, flags, _ => { }, socket);
   return Task.Factory.FromAsync(result,(r) => socket.EndSend(r));
}

现在事情变得简单了一点：

使用默认的BlockingCollection<> 作为并发队列。它是线程保存并删除列表上的显式锁定

static BlockingCollection<byte[]> buffer = new BlockingCollection<byte[]>();

public async void Main()
{
   Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
   socket.Connect(new IPEndPoint(new IPAddress(new byte[] { 192, 168, 1, 8 }), 123));
   while (!buffer.IsCompleted)
   {
      var data = buffer.Take();
      await SendAsync(socket, data, 0, data.Length, 0);
   }
   Console.ReadLine();            
}

在不需要信号量的同时保持非阻塞发送和顺序。