TPL 数据流反馈回路

Question

管道获取文件的绝对路径（Block 1）并对其进行处理并将其保存到数据库（block3）。

限制是某些文件类型 (.vde) 依赖于特定的父文件类型 (.vd)。父文件类型具有处理相关文件类型所需的元数据。除非系统中存在父文件类型，否则我无法处理从属文件类型。

目标 - 我需要系统以某种方式等待父文件类型进入系统并更新状态。然后会自动再次调用并处理依赖的文件类型。

我的方法 - 添加链接到 block 1 的反馈循环 (block 4)。 但是，我最终会丢失消息。来自block 4 到block 1 的消息无法到达block 3，并且在管道中的某个地方丢失了。

我该怎么做才能不丢失来自block 4 到block 1 的消息？

或者这可以用更好的方式完成吗？

Answer 1

作为练习，我尝试制作一个类似于JoinBlock<T1,T2> 的JoinDependencyBlock，它从两个缓冲区传播匹配的元素。

---

更新：我想出了一个更简单的实现，它在内部使用三个内置块，两个ActionBlocks 用于输入，一个BufferBlock 用于输出。每个动作块都会填充一个专用的List，当添加一个元素时，会在两个列表中搜索匹配的对。如果找到一个，则将其发布到BufferBlock。主要的复杂性在于这些块的链接，因为成功和失败的情况需要不同的处理。

JoinDependencyBlock 完成后，内部列表中所有不匹配的元素都将被丢弃。

public class JoinDependencyBlock<T1, T2> : ISourceBlock<(T1, T2)>
{
    private readonly Func<T1, T2, bool> _matchPredicate;
    private readonly List<T1> _list1 = new List<T1>();
    private readonly List<T2> _list2 = new List<T2>();
    private readonly ActionBlock<T1> _input1;
    private readonly ActionBlock<T2> _input2;
    private readonly BufferBlock<(T1, T2)> _output;
    private readonly object _locker = new object();

    public JoinDependencyBlock(Func<T1, T2, bool> matchPredicate,
        CancellationToken cancellationToken)
    {
        _matchPredicate = matchPredicate
            ?? throw new ArgumentNullException(nameof(matchPredicate));

        // Create the three internal blocks
        var options = new ExecutionDataflowBlockOptions()
        {
            CancellationToken = cancellationToken
        };
        _input1 = new ActionBlock<T1>(Add1, options);
        _input2 = new ActionBlock<T2>(Add2, options);
        _output = new BufferBlock<(T1, T2)>(options);

        // Link the input blocks with the output block
        var inputTasks = new Task[] { _input1.Completion, _input2.Completion };
        Task.WhenAny(inputTasks).Unwrap().ContinueWith(t =>
        {
            // If ANY input block fails, then the whole block has failed
            ((IDataflowBlock)_output).Fault(t.Exception.InnerException);
            if (!_input1.Completion.IsCompleted) _input1.Complete();
            if (!_input2.Completion.IsCompleted) _input2.Complete();
            ClearLists();
        }, default, TaskContinuationOptions.OnlyOnFaulted |
            TaskContinuationOptions.RunContinuationsAsynchronously,
            TaskScheduler.Default);
        Task.WhenAll(inputTasks).ContinueWith(t =>
        {
            // If ALL input blocks succeeded, then the whole block has succeeded
            _output.Complete();
            ClearLists();
        }, default, TaskContinuationOptions.NotOnFaulted |
            TaskContinuationOptions.RunContinuationsAsynchronously,
            TaskScheduler.Default);
    }

    public JoinDependencyBlock(Func<T1, T2, bool> matchPredicate)
        : this(matchPredicate, CancellationToken.None) { }

    public ITargetBlock<T1> Target1 => _input1;
    public ITargetBlock<T2> Target2 => _input2;
    public Task Completion => _output.Completion;

    private void Add1(T1 value1)
    {
        T2 value2;
        lock (_locker)
        {
            var index = _list2.FindIndex(v => _matchPredicate(value1, v));
            if (index < 0)
            {
                // Match not found
                _list1.Add(value1);
                return;
            }
            value2 = _list2[index];
            _list2.RemoveAt(index);
        }
        _output.Post((value1, value2));
    }

    private void Add2(T2 value2)
    {
        T1 value1;
        lock (_locker)
        {
            var index = _list1.FindIndex(v => _matchPredicate(v, value2));
            if (index < 0)
            {
                // Match not found
                _list2.Add(value2);
                return;
            }
            value1 = _list1[index];
            _list1.RemoveAt(index);
        }
        _output.Post((value1, value2));
    }

    private void ClearLists()
    {
        lock (_locker)
        {
            _list1.Clear();
            _list2.Clear();
        }
    }

    public void Complete() => _output.Complete();

    public void Fault(Exception exception)
        => ((IDataflowBlock)_output).Fault(exception);

    public IDisposable LinkTo(ITargetBlock<(T1, T2)> target,
        DataflowLinkOptions linkOptions)
        => _output.LinkTo(target, linkOptions);

    (T1, T2) ISourceBlock<(T1, T2)>.ConsumeMessage(
        DataflowMessageHeader messageHeader, ITargetBlock<(T1, T2)> target,
        out bool messageConsumed)
        => ((ISourceBlock<(T1, T2)>)_output).ConsumeMessage(
            messageHeader, target, out messageConsumed);

    void ISourceBlock<(T1, T2)>.ReleaseReservation(
        DataflowMessageHeader messageHeader, ITargetBlock<(T1, T2)> target)
        => ((ISourceBlock<(T1, T2)>)_output).ReleaseReservation(
            messageHeader, target);

    bool ISourceBlock<(T1, T2)>.ReserveMessage(
        DataflowMessageHeader messageHeader, ITargetBlock<(T1, T2)> target)
        => ((ISourceBlock<(T1, T2)>)_output).ReserveMessage(
            messageHeader, target);
}

使用示例：

var joinBlock = new JoinDependencyBlock<FileInfo, FileInfo>((fi1, fi2) =>
{
    // Check if the files are matched
    var name1 = Path.GetFileNameWithoutExtension(fi1.Name);
    var name2 = Path.GetFileNameWithoutExtension(fi2.Name);
    return StringComparer.OrdinalIgnoreCase.Equals(name1, name2);
});
var actionBlock = new ActionBlock<(FileInfo, FileInfo)>(pair =>
{
    // Process the matching files
    Console.WriteLine(pair.Item1.Name + " :: " +  pair.Item2.Name);
});
joinBlock.LinkTo(actionBlock);