c# - 超时立即取消阻塞操作

Question

我有一个从队列中读取的阻塞操作，但它可能会超时。我可以轻松地将其转换为“异步”操作:

    public async Task<IMessage> ReceiveAsync(CancellationToken cancellationToken)
    {
        return await Task.Run(() =>
        {
            while (true)
            {
                cancellationToken.ThrowIfCancellationRequested();

                // Try receiving for one second
                IMessage message = consumer.Receive(TimeSpan.FromSeconds(1));

                if (message != null)
                {
                    return message;
                }
            }
        }, cancellationToken).ConfigureAwait(false);
    }

中止线程通常被认为是不好的做法，因为您可能会泄漏资源，因此超时似乎是干净地停止线程的唯一方法。所以我有三个问题:

1. 普遍接受的“立即”取消超时值是多少？
2. 对于提供内置异步方法的库，立即取消是否真的存在，或者它们是否也使用超时和循环来模拟它？也许这里的问题是您将如何利用软件中断，以及这些中断是否还必须进行某种轮询以检查是否存在中断，即使是在内核/CPU 级别。
3. 有没有其他方法可以解决这个问题？

编辑:所以我可能已经通过 Thread.Interrupt() 找到了部分答案，然后处理了 ThreadInterruptedException。这基本上是一个内核级软件中断并且尽可能接近“立即”吗？以下是处理此问题的更好方法吗？

    public async Task<IMessage> ReceiveAsync(CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();

        var completionSource = new TaskCompletionSource<IMessage>();

        var receiverThread = new Thread(() =>
        {
            try
            {
                completionSource.SetResult(consumer.Receive());
            }
            catch (ThreadInterruptedException)
            {
                completionSource.SetCanceled();
            }
            catch (Exception ex)
            {
                completionSource.SetException(ex);
            }
        });

        cancellationToken.Register(receiverThread.Interrupt);

        receiverThread.Name = "Queue Receive";
        receiverThread.Start();

        return await completionSource.Task.ConfigureAwait(false);
    }

Answer 1

1. 这取决于您的具体需求。一秒钟对某些人来说可能很快，对其他人来说可能很慢。
2. 提供 async API 的库(好的库)自下而上地执行此操作。它们通常不会用线程包装阻塞(同步)操作以使其看起来异步。他们使用 TaskCompletionSource 来创建真正的async 方法。
3. 我不确定你所说的队列是什么意思(.Net 中内置的 Queue 没有 Receive 方法)但你可能应该使用一个真正的异步数据结构，如TPL Dataflow的BufferBlock。

关于您的特定代码示例。您在整个操作过程中都占用了一个线程(即 async over sync )，这是很昂贵的。您可以改为尝试快速消费，然后异步等待超时结束，或者等待 CancellationToken 被取消。在 Task.Run 中使用另一个线程也没有意义。您可以简单地将 async lambda 作为 ReceiveAsync 的内容:

public async Task<IMessage> ReceiveAsync(CancellationToken cancellationToken)
{
    while (true)
    {
        cancellationToken.ThrowIfCancellationRequested();
        // Try receiving for one second
        IMessage message;
        if (!consumer.TryReceive(out message))
        {
             await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
        }
        if (message != null)
        {
            return message;
        }
    }
}

如果您的队列实现了 IDisposable，一个不同的(更严厉的)选项是在 CancellationToken 被取消时对其调用 Dispose。 Here's how .