异步读取同步

Question

我似乎找不到一个优雅的解决方案来同步下面的代码。我记得曾经做过类似的事情，但现在我的大脑一片空白。

我需要保护没有两个线程会从套接字读取，始终只有一个线程等待 BeginRead。当我以这种方式使用 Mutex 时，会出现 SyncronizationError。

class Program
{
    TcpClient client = new TcpClient();
    public readonly object loker = new object();

    public void BeginRead()
    {
        if (!Mutex.TryEnter(loker))
        {
            return;
        }
        client.GetStream().BeginRead(..., Read_Callback);
    }

    private void Read_Callback(IAsyncResult ar) 
    {
        client.GetStream().EndRead(ar);
        Mutex.Exit(loker);
        BeginRead();
    }


    static void Main(string[] args)
    {
        BeginRead();
    }
}

Answer 1

您需要的是为异步而不是同步的关键部分设计的同步方法。您希望能够在代码能够取出锁定之后异步运行一些代码，而不是让您的异步方法使用同步锁定机制。这些同步机制通常也期望锁从被取出的同一个线程中释放；对于没有意义的异步操作。 SemaphoreSlim 专门设计用于支持异步同步。

private TcpClient client = new TcpClient();
private SemaphoreSlim semaphore = new SemaphoreSlim(1, 1);

public void BeginRead()
{
    semaphore.WaitAsync()
        .ContinueWith(t =>
            client.GetStream()
            .BeginRead(null, 0, 0, Read_Callback, null));
}

private void Read_Callback(IAsyncResult ar)
{
    client.GetStream().EndRead(ar);

    semaphore.Release();
    BeginRead();
}

您还可以选择使用 TPL 相当显着地重构此程序。

为此，您需要创建一个方法，该方法根据您感兴趣的异步操作生成Task，而不是另一种异步风格：

public static Task<int> WhenRead(
    this NetworkStream stream,
    byte[] buffer,
    int offset,
    int size)
{
    var tcs = new TaskCompletionSource<int>();
    stream.BeginRead(buffer, offset, size, result =>
    {
        tcs.TrySetResult(stream.EndRead(result));
    }, null);
    return tcs.Task;
}

但是一旦你有了这个程序，它就会变得更容易使用：

public async void BeginRead()
{
    while (true)
    {
        await semaphore.WaitAsync();
        int result = await client.GetStream().WhenRead(null, 0, 0);
        semaphore.Release();
        DoStuffWithResult(result);
    }
}

Answer 2

另一个解决方案是使用 ReaderWriterLockSlim。它有 IsReadLockHeld 方法。

Answer 3

首先，您将Monitor 类与Mutex 类混合在一起。 Mutex 没有这些静态方法。

现在，您正在经历的事实是，您在回调中使用了Monitor.Exit，这可能是从不同的线程调用的。这不起作用，因为必须从拥有锁对象的线程调用Monitor.Exit。

相反，您可以使用没有此要求的其他锁定原语，例如 Semaphore：

TcpClient client = new TcpClient();
private static readonly Semaphore Semaphore = new Semaphore(1, 1);

public void BeginRead()
{
    if (!Semaphore.WaitOne(0))
    {
        return;
    }
    client.GetStream().BeginRead(..., Read_Callback);
}

private void Read_Callback(IAsyncResult ar)
{
    client.GetStream().EndRead(ar);
    Semaphore.Release();
    BeginRead();
}

Answer 4

不能在一个线程上获取Mutex 并在另一个线程上释放。它必须由获取它的线程释放。互斥锁所有者是一个线程。

我强烈建议不要做任何你试图做的事情（在异步调用期间保持锁定）。一般模式是在发布调用和完成期间锁定：

lock(something) {
   ... 
   BeginIO(...)
}


completion:

   result = endIO;
   lock(something)
   {
     ...
   }