Использование Task.Run для запуска непрерывного фонового задания с BlockingCollection - PullRequest
Новая
       9

Использование Task.Run для запуска непрерывного фонового задания с BlockingCollection

0 голосов
/ 17 февраля 2019

Я использую Task.Run для запуска задания непрерывной фоновой обработки на BlockingCollection.Это работает, но я не думаю, что это идеальный способ структурировать код.

Эта проблема касается строки: public async Task ProcessQueue().Я получаю предупреждение:

This async method lacks 'await' operators and will run synchronously.

При удалении async возникает ошибка:

'MessageProcessor.ProcessQueue()': not all code paths return a value

Это разумно, потому что компилятор не видит, что getConsumingEnumerable блокируется.Должен ли я использовать версию async и игнорировать это предупреждение или реструктурировать свой код другим способом?

Примечание: представьте, что у меня есть 100 из этих MessageProcessor, я не думаю, что создаю отдельный потокдля каждого из них было бы эффективное использование памяти, следовательно, мое использование Task.Run. 

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public class Message
{
    public int id { get; set; }
    public int value { get; set; }
}

public class MessageProcessor
{
    private CancellationToken token;
    private BlockingCollection<Message> messageQueue;
    private List<Message> processedMessageList;

    public MessageProcessor(CancellationToken token)
    {
        this.token = token;
        messageQueue = new BlockingCollection<Message>(new ConcurrentQueue<Message>());
        processedMessageList = new List<Message>(10);
    }

    public void Add(Message m)
    {
        if (!messageQueue.IsAddingCompleted)
        {
            messageQueue.Add(m);
        }
    }

    public void CompleteAdding()
    {
        messageQueue.CompleteAdding();
    }

    public async Task ProcessQueue()
    {
        foreach (Message m in messageQueue.GetConsumingEnumerable())
        {
            Console.WriteLine($"Processing. id: {m.id}, value: {m.value}");
            processedMessageList.Add(m);

        }
    }
}

class Program
{
    static void Main(string[] args)
    {
        Random random = new Random();
        CancellationTokenSource tokenSource = new CancellationTokenSource();
        CancellationToken token = tokenSource.Token;
        MessageProcessor messageProcessor = new MessageProcessor(token);

        var messageProcessorResult = Task.Run(() => messageProcessor.ProcessQueue());

        for (int i = 0; i < 10; i++)
        {
            Message m = new Message();
            m.id = i;
            m.value = random.Next(-2, 2);

            Console.WriteLine($"Producing. id: {m.id}, value: {m.value}");
            messageProcessor.Add(m);
        }

        messageProcessor.CompleteAdding();
        messageProcessorResult.Wait();
    }
}

Ответы [ 2 ]

0 голосов
/ 17 февраля 2019

Вам нужен блок действий .

0 голосов
/ 17 февраля 2019

Удалите async и напишите следующее в конце метода ProcessQueue: 

return Task.CompletedTask;

Таким образом мы решим ошибку not all code paths return a value.

...