检测 DataFlow 中的重复项答案

【问题标题】：Detect Duplicate Items in DataFlow检测 DataFlow 中的重复项
【发布时间】：2017-08-07 16:48:29
【问题描述】：

我一直在构建一个服务，该服务使用Queue<string> 对象来处理文件以管理项目。

public partial class BasicQueueService : ServiceBase
{
    private readonly EventWaitHandle completeHandle = 
      new EventWaitHandle(false, EventResetMode.ManualReset, "ThreadCompleters");

    public BasicQueueService()
    {
        QueueManager = new Queue<string>();
    }

    public bool Stopping { get; set; }

    private Queue<string> QueueManager { get; }

    protected override void OnStart(string[] args)
    {
        Stopping = false;

        ProcessFiles();
    }

    protected override void OnStop()
    {
        Stopping = true;
    }

    private void ProcessFiles()
    {
        while (!Stopping)
        {
            var count = QueueManager.Count;
            for (var i = 0; i < count; i++)
            {
                //Check the Stopping Variable again.
                if (Stopping) break;

                var fileName = QueueManager.Dequeue();
                if (string.IsNullOrWhiteSpace(fileName) || !File.Exists(fileName)) 
                       continue;

                Console.WriteLine($"Processing {fileName}");

                Task.Run(() =>
                    {
                        DoWork(fileName);
                    })
                    .ContinueWith(ThreadComplete);
            }
            if (Stopping) continue;

            Console.WriteLine("Waiting for thread to finish, or 1 minute.");
            completeHandle.WaitOne(new TimeSpan(0, 0, 15));
            completeHandle.Reset();
        }
    }

    partial void DoWork(string fileName);

    private void ThreadComplete(Task task)
    {
        completeHandle.Set();
    }

    public void AddToQueue(string file)
    {
        //Called by FileWatcher/Manual classes, not included for brevity.
        lock (QueueManager)
        {
            if (QueueManager.Contains(file)) return;

            QueueManager.Enqueue(file);
        }
    }
}

在研究如何限制线程数量的同时（我尝试了一个手动类，其递增 int，但存在一个问题，即它在我的代码中没有正确递减），我遇到了 @987654321 @，这似乎更适合我想要实现的目标 - 具体来说，它允许我让框架处理线程/队列等。

现在这是我的服务：

public partial class BasicDataFlowService : ServiceBase
{
    private readonly ActionBlock<string> workerBlock;

    public BasicDataFlowService()
    {
        workerBlock = new ActionBlock<string>(file => DoWork(file), new ExecutionDataflowBlockOptions()
        {
            MaxDegreeOfParallelism = 32
        });
    }

    public bool Stopping { get; set; }

    protected override void OnStart(string[] args)
    {
        Stopping = false;
    }

    protected override void OnStop()
    {
        Stopping = true;
    }

    partial void DoWork(string fileName);

    private void AddToDataFlow(string file)
    {
        workerBlock.Post(file);
    }
}

这很好用。但是，我想确保一个文件只添加到TPL DataFlow 一次。使用Queue，我可以使用.Contains() 进行检查。有没有可以用于TPL DataFlow 的机制？

【问题讨论】：

无论是消费还是提交文件，都有责任不发布两次。如果您从目录中读取文件，您可以标记它们，或者按照@VMAtm 的建议缓存路径。但是，如果用户或其他客户正在提交它们，您需要将这些流程视为一项工作。其中每个文件代表具有单个结果的单个作业。

标签： c# multithreading tpl-dataflow dataflow

【解决方案1】：

您的Queue 解决方案仅在文件在短时间内两次进入您的服务时才有效。如果它在几个小时后再次出现，队列将不会包含它，因为你从那里Dequeue它。

如果需要此解决方案，那么您可以使用MemoryCache 来存储已处理的文件路径，如下所示：

using System.Runtime.Caching;

private static object _lock = new object();

private void AddToDataFlow(string file)
{
    lock (_lock)
    {
        if (MemoryCache.Default.Contains(file))
        {
            return;
        }

        // no matter what to put into the cache
        MemoryCache.Default[file] = true;
    // we can now exit the lock
    }

    workerBlock.Post(file);
}

但是，如果您的应用程序必须运行很长时间（打算执行哪个服务），您最终会耗尽内存。在这种情况下，您可能需要将文件路径存储在数据库或其他东西中，因此即使重新启动服务，您的代码也会恢复状态。

【讨论】：

【解决方案2】：

你可以在DoWork里面查看。

你必须保存在Hash已经工作的项目并检查当前文件名不存在于哈希中。

【讨论】：