0

我编写了一个小程序来测试使用 BufferBlock (System.Threading.Tasks.Dataflow) 来实现双优先级消费者-生产者队列。

消费者应始终首先使用高优先级队列中的任何项目。

在这个初始测试中，我让生产者运行的速度比消费者慢得多，所以数据应该按照它进入的顺序出来，不管优先级如何。

但是，我发现Task.WhenAny()直到两个队列中都有东西（或有完成），结果才完成，因此表现得像Task.WhenAll().

我以为我理解async/ await，并且我仔细阅读了 Cleary 的“C# Cookbook 中的并发性”。但是，发生了一些我不明白的事情。

有任何想法吗？

代码：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;  // add nuget package, 4.8.0
using static System.Console;

namespace DualBufferBlockExample { // .Net Framework 4.6.1

    class Program {
        private static async Task Produce(BufferBlock<int> queueLo, BufferBlock<int> queueHi, IEnumerable<int> values) {
            await Task.Delay(10);
            foreach(var value in values) {
                if(value == 3 || value == 7)
                    await queueHi.SendAsync(value);
                else
                    await queueLo.SendAsync(value);
                WriteLine($"Produced {value}  qL.Cnt={queueLo.Count} qH.Cnt={queueHi.Count}");
                await Task.Delay(1000);  // production lag
            }

            queueLo.Complete();
            queueHi.Complete();
        }
        private static async Task<IEnumerable<int>> Consume(BufferBlock<int> queueLo, BufferBlock<int> queueHi) {
            var results = new List<int>();

            while(true) {
                int value = -1;

                while(queueLo.Count > 0 || queueHi.Count > 0) {  // take from hi-priority first
                    if(queueHi.TryReceive(out value) ||
                        queueLo.TryReceive(out value)) {  // process value
                        results.Add(value);
                        WriteLine($"    Consumed {value}");
                        await Task.Delay(100); // consumer processing time shorter than production
                    }
                }

                var hasNorm = queueHi.OutputAvailableAsync();
                var hasLow = queueLo.OutputAvailableAsync();
                var anyT = await Task.WhenAny(hasNorm, hasLow);  // <<<<<<<<<< behaves like WhenAll
                WriteLine($"  WhenAny {anyT.Result} qL.Result={hasLow.Result} qH.Result={hasNorm.Result} qL.Count={queueLo.Count} qH.Count={queueHi.Count}");

                if(!anyT.Result)
                    break;  // both queues are empty & complete
            }

            return results;
        }
        static async Task TestDataFlow() {
            var queueLo = new BufferBlock<int>();
            var queueHi = new BufferBlock<int>();

            // Start the producer and consumer.
            var consumer = Consume(queueLo, queueHi);
            WriteLine("Consumer Started");

            var producer = Produce(queueLo, queueHi, Enumerable.Range(0, 10));
            WriteLine("Producer Started");

            // Wait for everything to complete.
            await Task.WhenAll(producer, consumer, queueLo.Completion, queueHi.Completion);

            // show consumer's output
            var results = await consumer;

            Write("Results:");
            foreach(var x in results)
                Write($" {x}");
            WriteLine();
        }
        static void Main(string[] args) {
            try {
                TestDataFlow().Wait();
            } catch(Exception ex) {
                WriteLine($"TestDataFlow exception: {ex.ToString()}");
            }
            ReadLine();
        }
    }
}

输出：

Consumer Started
Producer Started
Produced 0  qL.Cnt=1 qH.Cnt=0
Produced 1  qL.Cnt=2 qH.Cnt=0
Produced 2  qL.Cnt=3 qH.Cnt=0
Produced 3  qL.Cnt=3 qH.Cnt=1
  WhenAny True qL.Result=True qH.Result=True qL.Count=3 qH.Count=1
    Consumed 3
    Consumed 0
    Consumed 1
    Consumed 2
Produced 4  qL.Cnt=1 qH.Cnt=0
Produced 5  qL.Cnt=2 qH.Cnt=0
Produced 6  qL.Cnt=3 qH.Cnt=0
Produced 7  qL.Cnt=3 qH.Cnt=1
  WhenAny True qL.Result=True qH.Result=True qL.Count=3 qH.Count=1
    Consumed 7
    Consumed 4
    Consumed 5
    Consumed 6
Produced 8  qL.Cnt=1 qH.Cnt=0
Produced 9  qL.Cnt=2 qH.Cnt=0
  WhenAny True qL.Result=True qH.Result=False qL.Count=2 qH.Count=0
    Consumed 8
    Consumed 9
  WhenAny False qL.Result=False qH.Result=False qL.Count=0 qH.Count=0
Results: 3 0 1 2 7 4 5 6 8 9
4

1 回答 1

2

在调用WhenAny您立即阻止使用这两个任务后，.Result不知道它们都已完成。 

var anyT = await Task.WhenAny(hasNorm, hasLow);

//This line blocks on both the hasNorm and hasLow tasks preventing execution from continuing. 
WriteLine($"  WhenAny {anyT.Result} qL.Result={hasLow.Result} qH.Result={hasNorm.Result} qL.Count={queueLo.Count} qH.Count={queueHi.Count}");

awaiting这两个任务也会给你同样的行为。您可以做的最好的事情是从完成await的任务返回的任务，WhenAny并且只打印完成任务的结果。

此外，优先级队列并不是TPL-Dataflow开箱即用的。它平等地对待所有消息，因此您结束插入自己的优先级实现。那就是说你可以让它工作。

于 2018-02-07T16:03:04.320 回答