从生产环境者/消费者队列中删除取消的任务-Java 学习之路

我想使用异步生产环境者/消费者队列（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();
        }
    }
}

Edit 1:

正如svik指出的那样，只有在队列已经完成时才会抛出InvalidOperationException . 所以这个解决方案甚至没有解决我的等待任务的非托管“队列”的初始问题 . 如果有例如超过10次调用/ 10s我得到了一个完整的队列和一个额外的非托管“队列”等待任务，比如我的异步阻塞方法（AsyncMonitor） . 我想我必须提出一些其他的解决方案呢......

Edit 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条异步等待入队的消息，我不知道=> 生产环境者代码无论如何也无法正常工作，因为它需要永远发送等待的所有消息...）

我希望这能让我更清楚我想达到的目标;）

1 回答

我不确定回答我自己的任务是否正常，所以我不会将其标记为答案，也许有人想出一个更好的解决方案：P

首先，这里是生产环境者代码：

static async Task StartAsync()
{
    using (var queue = new SendMessageQueue(10, new SendMessageService()))
    using (var timeoutTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(4.5)))
    {
        var tasks = new List<Task>();

        for (var i = 0; i < 10; i++)
        {
            tasks.Add(queue.SendAsync(i.ToString(), timeoutTokenSource.Token));
        }
        await Task.Delay(TimeSpan.FromSeconds(4.5));
        for (var i = 10; i < 25; i++)
        {
            tasks.Add(queue.SendAsync(i.ToString(), default(CancellationToken)));
        }

        await queue.CompleteSendingAsync();

        for (var i = 0; i < tasks.Count; i++ )
        {
            try
            {
                await tasks[i];
                Console.WriteLine("Message '{0}' send.", i);
            }
            catch (TaskCanceledException)
            {
                Console.WriteLine("Message '{0}' canceled.", i);
            }
            catch (QueueOverflowException ex)
            {
                Console.WriteLine(ex.Message);
            }
        }
    }
    Console.WriteLine("Press any key to continue...");
    Console.ReadKey();
}

25条消息在5秒内排队
发送
16条消息
3条消息未发送（队列已满）
6条消息被取消

这里是基于List的“Queue”类 . 它是队列和消费者的组合 . 使用AsyncMonitor类（Stephen Cleary的AsyncEx）完成同步 .

class SendMessageQueue : IDisposable
{
    private bool m_Disposed;
    private bool m_CompleteSending;
    private Task m_SendingTask;
    private AsyncMonitor m_Monitor;
    private List<MessageTaskCompletionSource> m_MessageCollection;
    private ISendMessageService m_SendMessageService;

    public int Capacity { get; private set; }


    public SendMessageQueue(int capacity, ISendMessageService service)
    {
        Capacity = capacity;
        m_Monitor = new AsyncMonitor();
        m_MessageCollection = new List<MessageTaskCompletionSource>();
        m_SendMessageService = service;
        m_SendingTask = StartSendingAsync();
    }

    public async Task<bool> SendAsync(string message, CancellationToken token)
    {
        if (m_Disposed) { throw new ObjectDisposedException(GetType().Name); }
        if (message == null) { throw new ArgumentNullException("message"); }

        using (var messageTcs = new MessageTaskCompletionSource(message, token))
        {
            await AddAsync(messageTcs);
            return await messageTcs.Task;
        }
    }

    public async Task CompleteSendingAsync()
    {
        if (m_Disposed) { throw new ObjectDisposedException(GetType().Name); }

        using (m_Monitor.Enter())
        {
            m_CompleteSending = true;
        }
        await m_SendingTask;
    }

    private async Task AddAsync(MessageTaskCompletionSource message)
    {
        using (await m_Monitor.EnterAsync(message.Token))
        {
            if (m_CompleteSending) { throw new InvalidOperationException("Queue already completed."); }
            if (Capacity < m_MessageCollection.Count)
            {
                m_MessageCollection.RemoveAll(item => item.IsCanceled);
                if (Capacity < m_MessageCollection.Count)
                {
                    throw new QueueOverflowException(string.Format("Queue overflow; '{0}' couldn't be enqueued.", message.Message));
                }
            }
            m_MessageCollection.Add(message);
        }
        m_Monitor.Pulse(); // signal new message
        Console.WriteLine("Thread '{0}' - {1}: '{2}' enqueued.", Thread.CurrentThread.ManagedThreadId, DateTime.Now.TimeOfDay, message.Message);
    }

    private async Task<MessageTaskCompletionSource> TakeAsync()
    {
        using (await m_Monitor.EnterAsync())
        {
            var message = m_MessageCollection.ElementAt(0);
            m_MessageCollection.RemoveAt(0);
            return message;
        }
    }

    private async Task<bool> OutputAvailableAsync()
    {
        using (await m_Monitor.EnterAsync())
        {
            if (m_MessageCollection.Count > 0) { return true; }
            else if (m_CompleteSending) { return false; }

            await m_Monitor.WaitAsync();
            return true;
        }
    }

    private async Task StartSendingAsync()
    {
        while (await OutputAvailableAsync())
        {
            var message = await TakeAsync();
            if (message.IsCanceled) continue;
            try
            {
                var result = await m_SendMessageService.SendMessageAsync(message.Message, message.Token);
                message.TrySetResult(result);
            }
            catch (TaskCanceledException) { message.TrySetCanceled(); }
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected void Dispose(bool disposing)
    {
        if (m_Disposed) return;
        if (disposing)
        {
            if (m_MessageCollection != null)
            {
                var tmp = m_MessageCollection;
                m_MessageCollection = null;
                tmp.ForEach(item => item.Dispose());
                tmp.Clear();
            }
        }
        m_Disposed = true;
    }

    #region MessageTaskCompletionSource Class

    class MessageTaskCompletionSource : TaskCompletionSource<bool>, IDisposable
    {
        private bool m_Disposed;
        private IDisposable m_CancellationTokenRegistration;

        public string Message { get; private set; }
        public CancellationToken Token { get; private set; }
        public bool IsCompleted { get { return Task.IsCompleted; } }
        public bool IsCanceled { get { return Task.IsCanceled; } }
        public bool IsFaulted { get { return Task.IsFaulted; } }


        public MessageTaskCompletionSource(string message, CancellationToken token)
        {
            m_CancellationTokenRegistration = token.Register(() => TrySetCanceled());
            Message = message;
            Token = token;
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected void Dispose(bool disposing)
        {
            if (m_Disposed) return;
            if (disposing)
            {
                TrySetException(new ObjectDisposedException(GetType().Name));

                if (m_CancellationTokenRegistration != null)
                {
                    var tmp = m_CancellationTokenRegistration;
                    m_CancellationTokenRegistration = null;
                    tmp.Dispose();
                }
            }
            m_Disposed = true;
        }
    }

    #endregion
}

现在我对这个解决方案没问题;它完成了工作：D

回复于 2024-04-29T13:37:56+08:00

从 生产环境 者/消费者队列中删除取消的任务

1 回答

相关问题

从生产环境者/消费者队列中删除取消的任务