c# - TPL DataFlow，优先链接 block ？

Question

使用 TPL.DataFlow block ，是否可以将两个或多个源链接到单个 ITargetBlock(例如 ActionBlock)并确定源的优先级？

例如

BufferBlock<string> b1 = new ...
BufferBlock<string> b2 = new ...
ActionBlock<string> a = new ...

//somehow force messages in b1 to be processed before any message of b2, always
b1.LinkTo (a);
b2.LinkTo (a);

只要 b1 中有消息，我就希望将它们提供给“a”，一旦 b1 为空，b2 消息就会被推送到“a”

想法？

Answer 1

TPL 数据流本身没有类似的东西。

我能想到的最简单的方法是创建一个封装三个 block 的结构:高优先级输入、低优先级输入和输出。这些 block 将是简单的 BufferBlock，以及根据优先级将消息从两个输入转发到输出的方法，在后台运行。

代码可能是这样的:

public class PriorityBlock<T>
{
    private readonly BufferBlock<T> highPriorityTarget;

    public ITargetBlock<T> HighPriorityTarget
    {
        get { return highPriorityTarget; }
    }

    private readonly BufferBlock<T> lowPriorityTarget;

    public ITargetBlock<T> LowPriorityTarget
    {
        get { return lowPriorityTarget; }
    }

    private readonly BufferBlock<T> source;

    public ISourceBlock<T> Source
    {
        get { return source; }
    }

    public PriorityBlock()
    {
        var options = new DataflowBlockOptions { BoundedCapacity = 1 };

        highPriorityTarget = new BufferBlock<T>(options);
        lowPriorityTarget = new BufferBlock<T>(options);
        source = new BufferBlock<T>(options);

        Task.Run(() => ForwardMessages());
    }

    private async Task ForwardMessages()
    {
        while (true)
        {
            await Task.WhenAny(
                highPriorityTarget.OutputAvailableAsync(),
                lowPriorityTarget.OutputAvailableAsync());

            T item;

            if (highPriorityTarget.TryReceive(out item))
            {
                await source.SendAsync(item);
            }
            else if (lowPriorityTarget.TryReceive(out item))
            {
                await source.SendAsync(item);
            }
            else
            {
                // both input blocks must be completed
                source.Complete();
                return;
            }
        }
    }
}

用法看起来像这样:

b1.LinkTo(priorityBlock.HighPriorityTarget);
b2.LinkTo(priorityBlock.LowPriorityTarget);
priorityBlock.Source.LinkTo(a);

为此，a 还必须将 BoundingCapacity 设置为 1(或至少是一个非常低的数字)。

此代码需要注意的是，它可能会引入两条消息的延迟(一条在输出 block 中等待，一条在 SendAsync() 中等待)。因此，如果您有一长串低优先级消息，突然收到一条高优先级消息，则只有在这两条已经在等待的低优先级消息之后才会处理该消息。

如果这对您来说是个问题，它可以得到解决。但我相信它需要更复杂的代码，处理 TPL 数据流中不太公开的部分，比如 OfferMessage() .