从生产者/消费者队列中删除取消的任务(Remove cancelled Task from producer/consumer queue)

 我想使用异步生产者/消费者队列（AsyncEx lib）通过总线一次发送一条消息。 现在我只是通过异步阻塞来实现这一点。 它工作正常，但我无法控制队列:( 
 
 所以我想出了以下解决方案，问题是没有从队列中删除已取消的任务。 如果我将队列限制为10（因为每个消息需要1s发送，最大队列时间应为10s左右）并且队列包含8个等待任务和2个已取消任务，则下一个排队任务将抛出InvalidOperationException，尽管无论如何都不会发送两个被取消的任务。 
 
 也许有更好的方法来做到这一点：D 
 
    class Program
{
    static AsyncProducerConsumerQueue<Tuple<string, TaskCompletionSource>> s_Queue =
        new AsyncProducerConsumerQueue<Tuple<string, TaskCompletionSource>>();

    static void Main()
    {
        StartAsync().Wait();
    }

    static async Task StartAsync()
    {
        var sendingTask = StartSendingAsync();
        var tasks = new List<Task>();

        using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(8)))
        {
            for (var i = 0; i < 10; i++)
            {
                tasks.Add(EnqueueMessageAsync("Message " + i, cts.Token));
            }

            try
            {
                await Task.WhenAll(tasks);
                Console.WriteLine("All messages sent.");
            }
            catch (TaskCanceledException)
            {
                Console.WriteLine("At least one task was canceled.");
            }                
        }

        s_Queue.CompleteAdding();
        await sendingTask;
        s_Queue.Dispose();
        Console.WriteLine("Queue completed.");

        Console.WriteLine("Press any key to continue...");
        Console.ReadKey();
    }

    static async Task EnqueueMessageAsync(string message, CancellationToken token)
    {

        var tcs = new TaskCompletionSource();
        using (token.Register(() => tcs.TrySetCanceled()))
        {
            await s_Queue.EnqueueAsync(new Tuple<string, TaskCompletionSource>(message, tcs));
            Console.WriteLine("Thread '{0}' - {1}: {2} queued.", Thread.CurrentThread.ManagedThreadId, DateTime.Now.TimeOfDay, message);
            await tcs.Task;
        }
    }

    static async Task SendMessageAsync(string message)
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        Console.WriteLine("Thread '{0}' - {1}: {2} sent.", Thread.CurrentThread.ManagedThreadId, DateTime.Now.TimeOfDay, message);
    }

    static async Task StartSendingAsync()
    {
        while (await s_Queue.OutputAvailableAsync())
        {
            var t = await s_Queue.DequeueAsync();
            if (t.Item2.Task.IsCanceled || t.Item2.Task.IsFaulted) continue;

            await SendMessageAsync(t.Item1);
            t.Item2.TrySetResult();
        }
    }
}
 
 编辑1： 
 
 正如svik指出的那样，只有在队列已经完成时才会抛出InvalidOperationException。 所以这个解决方案甚至没有解决我的等待任务的非托管“队列”的初始问题。 如果有超过10个调用/ 10s，我得到一个完整的队列和一个额外的非托管“队列”等待任务，就像我的异步阻塞方法（AsyncMonitor）。 我想我必须提出一些其他的解决方案...... 
 
 编辑2： 
 
 我有N个不同的消息生成器（我不知道有多少消息因为它不是我的代码）而且只有一个消费者通过总线发送消息并检查它们是否被正确发送（不是真正的字符串消息）。 
 
 以下代码模拟了代码应该中断的情况（队列大小为10）： 
 
 
 
 排队10条消息（超时为5秒） 
 
 
 等待5秒（发送消息0-4，取消消息5-9） 
 
 
 排队11条新消息（没有超时） 
 
 
 消息10-19应该排队，因为队列只包含已取消的消息 
 
 
 消息20应抛出异常（例如QueueOverflowException），因为队列已满，生成器代码将处理或不处理 
 
 
 制片人： 
 
using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)))
{
    for (var i = 0; i < 10; i++) { tasks.Add(EnqueueMessageAsync("Message " + i, cts.Token)); }
    await Task.Delay(TimeSpan.FromSeconds(5));
    for (var i = 10; i < 21; i++) { tasks.Add(EnqueueMessageAsync("Message " + i, default(CancellationToken))); }

    try
    {
        await Task.WhenAll(tasks);
        Console.WriteLine("All messages sent.");
    }
    catch (TaskCanceledException)
    {
        Console.WriteLine("At least one task was canceled.");
        Console.WriteLine("Press any key to complete queue...");
        Console.ReadKey();
    }
}
 
 我的目标是，我想完全控制应该发送的所有消息，但在我之前发布的代码中并非如此，因为我只能控制队列中的消息而不是消息等待入队（可能有10000条异步等待入队的消息，我不知道=>生产者代码无论如何也无法正常工作，因为它需要永远发送所有等待的消息......） 
 
 我希望这能让我更清楚我想达到的目标;） 

I want to use an async producer/consumer queue (AsyncEx lib) to send messages one at a time over a bus. Right now I achieve this simply by async blocking. It's working fine, but I have no control over the queue :(
 
So I came up with following solution, problem is that a canceled task is not removed from the queue. If I limit the queue to say 10 (because each message takes 1s to send and max queue time shall be 10s or so) and the queue contains already 8 waiting tasks and 2 canceled tasks, than the next queued task would throw an InvalidOperationException although the two canceled task wouldn't be sent anyway.
 
Maybe there is a better way to do this :D
 
    class Program
{
    static AsyncProducerConsumerQueue<Tuple<string, TaskCompletionSource>> s_Queue =
        new AsyncProducerConsumerQueue<Tuple<string, TaskCompletionSource>>();

    static void Main()
    {
        StartAsync().Wait();
    }

    static async Task StartAsync()
    {
        var sendingTask = StartSendingAsync();
        var tasks = new List<Task>();

        using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(8)))
        {
            for (var i = 0; i < 10; i++)
            {
                tasks.Add(EnqueueMessageAsync("Message " + i, cts.Token));
            }

            try
            {
                await Task.WhenAll(tasks);
                Console.WriteLine("All messages sent.");
            }
            catch (TaskCanceledException)
            {
                Console.WriteLine("At least one task was canceled.");
            }                
        }

        s_Queue.CompleteAdding();
        await sendingTask;
        s_Queue.Dispose();
        Console.WriteLine("Queue completed.");

        Console.WriteLine("Press any key to continue...");
        Console.ReadKey();
    }

    static async Task EnqueueMessageAsync(string message, CancellationToken token)
    {

        var tcs = new TaskCompletionSource();
        using (token.Register(() => tcs.TrySetCanceled()))
        {
            await s_Queue.EnqueueAsync(new Tuple<string, TaskCompletionSource>(message, tcs));
            Console.WriteLine("Thread '{0}' - {1}: {2} queued.", Thread.CurrentThread.ManagedThreadId, DateTime.Now.TimeOfDay, message);
            await tcs.Task;
        }
    }

    static async Task SendMessageAsync(string message)
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        Console.WriteLine("Thread '{0}' - {1}: {2} sent.", Thread.CurrentThread.ManagedThreadId, DateTime.Now.TimeOfDay, message);
    }

    static async Task StartSendingAsync()
    {
        while (await s_Queue.OutputAvailableAsync())
        {
            var t = await s_Queue.DequeueAsync();
            if (t.Item2.Task.IsCanceled || t.Item2.Task.IsFaulted) continue;

            await SendMessageAsync(t.Item1);
            t.Item2.TrySetResult();
        }
    }
}
 
Edit 1:
 
As svik pointed out the InvalidOperationException is only thrown if the queue is already completed. So this solution doesn't even solve my initial problem of an unmanaged "queue" of waiting tasks. If there are e.g. more than 10 calls/10s I got a full queue and an additional unmanaged "queue" of waiting tasks like with my async blocking approach (AsyncMonitor). I guess I have to come up with some other solution then...
 
Edit 2:
 
I have N different producers of messages (I don't know how many there are because it's not my code) and only one consumer that sends the messages over a bus and checks if they were sent correctly (not really string messages). 
 
The following code simulates a situation where the code should break (queue size is 10):
 
 
 
Enqueue 10 messages (with an timeout of 5sec)
 
 
Wait 5sec (message 0-4 were sent and message 5-9 were cancelled)
 
 
Enqueue 11 new messages (w/o timeout)
 
 
Message 10 - 19 should be enqueued because the queue only contains cancelled messages
 
 
Message 20 should throw an exception (e.g. QueueOverflowException) because the queue is full, this would be handled or not by the producer code
 
 
Producers:
 
using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)))
{
    for (var i = 0; i < 10; i++) { tasks.Add(EnqueueMessageAsync("Message " + i, cts.Token)); }
    await Task.Delay(TimeSpan.FromSeconds(5));
    for (var i = 10; i < 21; i++) { tasks.Add(EnqueueMessageAsync("Message " + i, default(CancellationToken))); }

    try
    {
        await Task.WhenAll(tasks);
        Console.WriteLine("All messages sent.");
    }
    catch (TaskCanceledException)
    {
        Console.WriteLine("At least one task was canceled.");
        Console.WriteLine("Press any key to complete queue...");
        Console.ReadKey();
    }
}
 
The goal is, I want to have full control over all messages that should be send, but this is not the case in the code I've posted before, because I only have control over the messages in the queue but not the messages that are waiting to be enqueued (there could be 10000 messages asynchronously waiting to be enqueued and I wouldn't know => producer code wouldn't work as expected anyway because it would take forever to send all the messages that are waiting...)
 
I hope this makes it clearer what I want to achieve ;)

原文：https://stackoverflow.com/questions/29988340