c# - BlockingCollection 与异步任务

Question

我正在尝试正确建模多线程单生产者/多消费者场景，其中消费者可以要求生产者获取元素，但生产者需要执行耗时的操作来生产它(想想执行查询或打印文档)。

我的目标是确保没有消费者可以同时要求生产者生产一件商品。在我的真实世界用例中，生产者是一个硬件 Controller ，它必须确保一次只向硬件发送一个请求。其他并发请求最终必须等待或被拒绝(我知道如何拒绝它们，所以让我们集中精力让它们等待)。

我希望生产者和每个消费者在不同的线程中运行。
我无法仅使用 BlockingCollection 获得干净的代码。我必须将它与 SemaphoreSlim 一起使用，否则消费者可能会出现竞争条件。
我认为它应该有效(事实上它在我的所有测试中都有效)，即使我不是 100% 确定它。
这是我的程序:

制作人:

class Producer : IDisposable
{
    //Explicit waiting item => I feel this should not be there
    private SemaphoreSlim _semaphore;

    private BlockingCollection<Task<string>> _collection;

    public Producer()
    {
        _collection = new BlockingCollection<Task<string>>(new ConcurrentQueue<Task<string>>(), 1);
        _semaphore = new SemaphoreSlim(1, 1);
    }

    public void Start()
    {
        Task consumer = Task.Factory.StartNew(() =>
        {
            try
            {
                while (!_collection.IsCompleted)
                {
                    Task<string> current = _collection.Take();
                    current.RunSynchronously(); //Is this bad?

                    //Signal the long running operation has ended => This is what I'm not happy about
                    _semaphore.Release();
                }
            }
            catch (InvalidOperationException)
            {
                Console.WriteLine("Adding was compeleted!");
            }
        });
    }

    public string GetRandomString(string consumerName)
    {
        Task<string> task = new Task<string>(() =>
        {
            //Simulate long running operation
            Thread.Sleep(100);
            return GetRandomString();
        });

        _collection.Add(task);

        //Wait for long running operation to complete => This is what I'm not happy about
        _semaphore.Wait();

        Console.WriteLine("Producer produced {0} by {1} request", task.Result, consumerName);
        return task.Result;
    }

    public void Dispose()
    {
        _collection.CompleteAdding();
    }

    private string GetRandomString()
    {
        var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
        var random = new Random();
        var result = new string(Enumerable
            .Repeat(chars, 8)
            .Select(s => s[random.Next(s.Length)])
            .ToArray());
        return result;
    }
}

消费者:

class Consumer
{
    Producer _producer;
    string _name;

    public Consumer(
        Producer producer,
        string name)
    {
        _producer = producer;
        _name = name;
    }

    public string GetOrderedString()
    {
        string produced = _producer.GetRandomString(_name);
        return String.Join(String.Empty, produced.OrderBy(c => c));
    }
}

控制台应用程序:

class Program
{
    static void Main(string[] args)
    {
        int consumerNumber = 5;
        int reps = 10;

        Producer prod = new Producer();
        prod.Start();

        Task[] consumers = new Task[consumerNumber];

        for (var cConsumers = 0; cConsumers < consumerNumber; cConsumers++)
        {
            Consumer consumer = new Consumer(prod, String.Format("Consumer{0}", cConsumers + 1));

            Task consumerTask = Task.Factory.StartNew((consumerIndex) =>
            {
                int cConsumerNumber = (int)consumerIndex;
                for (var counter = 0; counter < reps; counter++)
                {
                    string data = consumer.GetOrderedString();
                    Console.WriteLine("Consumer{0} consumed {1} at iteration {2}", cConsumerNumber, data, counter + 1);
                }
            }, cConsumers + 1);

            consumers[cConsumers] = consumerTask;
        }

        Task continuation = Task.Factory.ContinueWhenAll(consumers, (c) =>
        {
            prod.Dispose();
            Console.WriteLine("Producer/Consumer ended");
            Console.ReadLine();
        });

        continuation.Wait();
    }
}

我关心的是这是否是解决问题的正确方法，或者你们是否可以推荐任何其他最佳实践。
我已经用谷歌搜索并尝试了不同的建议想法，但我尝试过的每个例子都假设生产者能够在他们被要求后立即生产元素……在现实世界的应用程序中这是非常罕见的情况:)
非常感谢任何帮助。

Answer 1

如果我对您的理解正确，您希望确保所谓的“生产者”一次只处理一项任务。然后对您的代码稍作修改，您可以这样做:

internal class Producer : IDisposable {
    private readonly BlockingCollection<RandomStringRequest> _collection;

    public Producer() {
        _collection = new BlockingCollection<RandomStringRequest>(new ConcurrentQueue<RandomStringRequest>());
    }

    public void Start() {
        Task consumer = Task.Factory.StartNew(() => {
            try {
                foreach (var request in _collection.GetConsumingEnumerable()) {
                    Thread.Sleep(100); // long work
                    request.SetResult(GetRandomString());
                }
            }
            catch (InvalidOperationException) {
                Console.WriteLine("Adding was compeleted!");
            }
        });
    }

    public RandomStringRequest GetRandomString(string consumerName) {
        var request = new RandomStringRequest();
        _collection.Add(request);
        return request;            
    }

    public void Dispose() {
        _collection.CompleteAdding();
    }

    private string GetRandomString() {
        var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
        var random = new Random();
        var result = new string(Enumerable
            .Repeat(chars, 8)
            .Select(s => s[random.Next(s.Length)])
            .ToArray());
        return result;
    }
}

internal class RandomStringRequest : IDisposable {
    private string _result;
    private ManualResetEvent _signal;

    public RandomStringRequest() {
        _signal = new ManualResetEvent(false);
    }

    public void SetResult(string result) {
        _result = result;
        _signal.Set();
    }

    public string GetResult() {
        _signal.WaitOne();
        return _result;
    }

    public bool TryGetResult(TimeSpan timeout, out string result) {
        result = null;
        if (_signal.WaitOne(timeout)) {
            result = _result;
            return true;
        }
        return false;
    }

    public void Dispose() {
        _signal.Dispose();
    }
}

internal class Consumer {
    private Producer _producer;
    private string _name;

    public Consumer(
        Producer producer,
        string name) {
        _producer = producer;
        _name = name;
    }

    public string GetOrderedString() {
        using (var request = _producer.GetRandomString(_name)) {
            // wait here for result to be prepared
            var produced = request.GetResult();
            return String.Join(String.Empty, produced.OrderBy(c => c));
        }
    }
}

请注意，生产者是单线程的，它使用 GetConsumingEnumerable。也没有信号量，也没有任务。相反，RandomStringRequest 返回给消费者，并且在调用 GetResult 或 TryGetResult 时，它将等待生产者产生结果(或超时到期)。您可能还想在某些地方传递 CancellationTokens(例如 GetConsumingEnumerable)。