c# - 允许线程有序进入的WaitHandle.WaitAny

Question

我有固定数量的“浏览器”，每个浏览器都不是线程安全的，因此必须在单个线程上使用。另一方面，我有一长串线程等待使用这些浏览器。我目前正在做的是有一个 AutoResetEvent 数组:

public readonly AutoResetEvent[] WaitHandles;

然后像这样初始化它们:

WaitHandles = Enumerable.Range(0, Browsers.Count).Select(_ => new AutoResetEvent(true)).ToArray();

所以我为每个浏览器设置了一个 AutoResetEvent，它允许我为每个线程检索一个特定的浏览器索引:

public Context WaitForBrowser(int i)
{
    System.Diagnostics.Debug.WriteLine($">>> WILL WAIT: {i}");
    var index = WaitHandle.WaitAny(WaitHandles);
    System.Diagnostics.Debug.WriteLine($">>> ENTERED: {i}");
    return new Context(Browsers[index], WaitHandles[index]);
}

这里的i只是等待线程的索引，因为这些线程在一个列表中并且有特定的顺序。我只是为了调试目的而传递它。 Context 是一次性的，然后在处理时调用等待句柄上的 Set。

当我查看我的输出时，我看到我所有的“>>> WILL WAIT: {i}”消息都是按正确顺序排列的，因为对 WaitForBrowser 的调用是按顺序进行的，但是我的“>>> ENTERED: {i}"消息的顺序是随机的(前几条除外)，因此它们的进入顺序与到达 var index = WaitHandle.WaitAny(WaitHandler); 行。

所以我的问题是，是否有任何方法可以修改它，以便线程以调用 WaitForBrowser 方法的相同顺序进入(这样“>>> ENTERED: {i}”消息是也订购)？

Answer 1

由于似乎没有开箱即用的解决方案，我最终使用了 this solution 的修改版本:

public class SemaphoreQueueItem<T> : IDisposable
{
    private bool Disposed;
    private readonly EventWaitHandle WaitHandle;
    public readonly T Resource;

    public SemaphoreQueueItem(EventWaitHandle waitHandle, T resource)
    {
        WaitHandle = waitHandle;
        Resource = resource;
    }

    public void Dispose()
    {
        if (!Disposed)
        {
            Disposed = true;
            WaitHandle.Set();
        }
    }
}

public class SemaphoreQueue<T> : IDisposable
{
    private readonly T[] Resources;
    private readonly AutoResetEvent[] WaitHandles;
    private bool Disposed;
    private ConcurrentQueue<TaskCompletionSource<SemaphoreQueueItem<T>>> Queue = new ConcurrentQueue<TaskCompletionSource<SemaphoreQueueItem<T>>>();

    public SemaphoreQueue(T[] resources)
    {
        Resources = resources;
        WaitHandles = Enumerable.Range(0, resources.Length).Select(_ => new AutoResetEvent(true)).ToArray();
    }

    public SemaphoreQueueItem<T> Wait(CancellationToken cancellationToken)
    {
        return WaitAsync(cancellationToken).Result;
    }

    public Task<SemaphoreQueueItem<T>> WaitAsync(CancellationToken cancellationToken)
    {
        var tcs = new TaskCompletionSource<SemaphoreQueueItem<T>>();
        Queue.Enqueue(tcs);

        Task.Run(() => WaitHandle.WaitAny(WaitHandles.Concat(new[] { cancellationToken.WaitHandle }).ToArray())).ContinueWith(task =>
        {
            if (Queue.TryDequeue(out var popped))
            {
                var index = task.Result;

                if (cancellationToken.IsCancellationRequested)
                    popped.SetResult(null);
                else
                    popped.SetResult(new SemaphoreQueueItem<T>(WaitHandles[index], Resources[index]));
            }
        });

        return tcs.Task;
    }

    public void Dispose()
    {
        if (!Disposed)
        {
            foreach (var handle in WaitHandles)
                handle.Dispose();

            Disposed = true;
        }
    }
}