TPL C# Windows 服务 - 有什么问题吗？

Question

您好用 TPL 编写 Windows 服务，要求是 Windows 服务应该每隔几分钟或几秒运行一次，并并行执行 4 个任务。我已经实现了以下代码，但有一个疑问，比如

Task[] taskRunners = new Task[Utility.conTaskRunners];

        Log("Stating Initiate Tasks::" + DateTime.Now);

        Log("Calling InitiateTaskRunner Tasks::" + DateTime.Now);

        taskRunners[0] = Task.Factory.StartNew(() => InitiateTaskRunner());
        taskRunners[1] = Task.Factory.StartNew(() => ScanTaskRunner());
        taskRunners[2] = Task.Factory.StartNew(() => ConvertTaskRunner());
        taskRunners[3] = Task.Factory.StartNew(() => ConvertedScanRunner());
        Task.WaitAll(taskRunners);
        Thread.Sleep(12000);

        InitiateTasks();

InitiateTaskRunner 如下所示

 public void InitiateTaskRunner()
    {
        Log("Begening InitiateTaskRunner ::" + DateTime.Now);           
        string[] subDirs = Directory.GetDirectories(Utility.inputPath);
        if (subDirs.Length > 0) //files eixsts go and process it
        {
            Parallel.ForEach(subDirs, dir =>
            {
                InitiateTaskWorker(dir);
            });
        }           
        //BEFORE I SET signal SHOULD this wait as above is PARALLEL.FOREACH?
        _waitOnOne.Set(); //set signal for successor after finish
        Log("Ending InitiateTaskRunner ::" + DateTime.Now);
    }

类似的其他 3 个 TaskRunners 将是；基本上每个任务运行器都会为每个子目录调用TaskWorker并异步处理文件并吃掉TASK Runner，每个Task Worker将是独立和并发的。 TaskWorker 处理完文件后，文件从一个 taskWorker 到另一个 Task Worker 位置将很快移动，并且下一个 TaskWorker 将拾取，但不会从一个 TaskWorker 触发到另一个 TaskWorker。

这种方法的问题是，有时 2、3 和 4 个任务运行器可能会在不执行任何操作的情况下退出，而第一个任务运行器仍在处理并推送到 taskrunner#2 位置等等。并且来自任务运行器位置 2,3,4 的文件将在下一次迭代中处理。

有没有更好的设计呢？并且任何有关 TPL 中异常处理的帮助都会很棒。

public void ScanTaskRunner()
    {
        Log("Begening ScanTaskRunner ::" + DateTime.Now);            
        string[] subDirs = Directory.GetDirectories(Utility.scanPath);
        if (subDirs.Length > 0)  //files exists so proceed
        {
            Parallel.ForEach(subDirs, dir =>
            {
                ScanTaskWorker(dir, false);
            });
        }
        else //no files hence wait for prcedent TASK to complete
        {
            _waitOnOne.WaitOne();
            Parallel.ForEach(Directory.GetDirectories(Utility.scanPath), dir =>
            {
                ScanTaskWorker(dir, false);
            });
        }
        _waitOnTwo.Set(); //Go successor if waiting for this
        Log("Ending ScanTaskRunner ::" + DateTime.Now);
    }
public void ConvertTaskRunner()
    {
       Log("Begening ConvertTasktRunner ::" + DateTime.Now);
        string[] subDirs = Directory.GetDirectories(Utility.convertPath);
        if (subDirs.Length > 0)  //files exists so process as usual
        {
            Parallel.ForEach(subDirs, dir =>
            {
                ConvertTaskWorker(dir);
            });
        }
        else  //no files hence wait for precedent TASK to complete
        {
            _waitOnTwo.WaitOne();
            Parallel.ForEach(Directory.GetDirectories(Utility.scanPath), dir =>
            {
                ScanTaskWorker(dir, false);
            });
        }
        _waitOnThree.Set();  //Successor is good to go if waiting
        Log("Ending ConvertTasktRunner ::" + DateTime.Now);
    }
 public void ConvertedScanRunner()
    {
        Log("Begening ConvertedScanRunner ::" + DateTime.Now);
        string[] subDirs = Directory.GetDirectories(Utility.convrtedScanPath);
        if (subDirs.Length > 0)  //files avaiable process as usual
        {
            Parallel.ForEach(subDirs, dir =>
            {
                ConvertedScanTaskWorker(dir);
            });
        }
        else  //no files wait until precendent TASK completes
        {
            _waitOnThree.WaitOne();
            Parallel.ForEach(Directory.GetDirectories(Utility.scanPath), dir =>
            {
                ScanTaskWorker(dir, false);
            });
        }
        Log("Ending ConvertedScanRunner ::" + DateTime.Now);
    }

Answer 1

请参阅Best way to do a task looping in Windows Service，了解启动任务并继续执行的好方法。

就异常处理等而言，您可能希望等待所有正在运行的任务，然后捕获并处理AggregateException。任何引发异常的任务都将具有IsFaulted true。在每个任务中使用ThrowIfCancellationRequested与其他所有人合作并参与被取消。那些合作的任务将有他们的IsCanceled 属性为真。

    public void Run(CancellationTokenSource cts)
    {
        _cts = cts;
        Task[] tasks = new Task[4];

        try
        {
            tasks[0] = Task.Factory.StartNew(() => PhaseOne(), cts.Token);
            tasks[1] = Task.Factory.StartNew(() => PhaseTwo(), cts.Token);
            tasks[2] = Task.Factory.StartNew(() => PhaseThree(), cts.Token);
            tasks[3] = Task.Factory.StartNew(() => PhaseFour(), cts.Token);

            Task.WaitAll(tasks);
        }
        catch (AggregateException ex)
        {
            // Handle exceptions
            foreach(Exception inner in ex.InnerExceptions)
            {
                // do something
            }
        }

        if (tasks.Any(t => t.IsFaulted))
        {
            // do something
        }
        if (tasks.Any(t => t.IsCanceled))
        {
            // do something
        }
    }

至于保持任务同步以及知道何时处理文件或何时完成另一个阶段的问题，您可以使用EventWaitHandle（尽管您可以通过多种方式做到这一点）。等待句柄可能比其他一些替代方案（如自旋锁）慢一点，因为它进入内核，但对于等待文件处理，使用它可能是一个好主意。

此处的代码显示了 PhaseTwo 的外观。 PhaseOne 看起来基本相同，但您会为一个不同的处理委托等使用等待句柄。

    private EventWaitHandle _waitOnTwo = new EventWaitHandle(false, EventResetMode.AutoReset);
    public void PhaseTwo()
    {
        try
        {
            string phaseTwoDir = "C:\\PhaseTwo";
            string phaseThreeDir = "C:\\PhaseThree";

            Action<string> processFiles = new Action<string>((file) => { File.Move(file, Path.Combine(phaseThreeDir, Path.GetFileName(file))); _waitOnTwo.Set(); });
            Func<bool> continueCondition = new Func<bool>(() => { return (Directory.GetFiles(phaseTwoDir).Count() > 0 || _waitOnOne.WaitOne(60000)); });

            // process files
            Process(phaseTwoDir, processFiles, continueCondition);
        }
        finally
        {
            _waitOnTwo.Set();
        }
    }

    private void Process(string directory, Action<string> processFileWork, Func<bool> continueCondition)
    {
        while (continueCondition())
        {
            foreach (string file in System.IO.Directory.GetFiles(directory))
            {
                // if canceled, throw
                _cts.Token.ThrowIfCancellationRequested();

                // process file
                processFileWork(file);
            }

        }
    }