c# - 排队调用套接字的异步方法 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();            
}

在不需要信号量的情况下保持非阻塞发送和顺序。