使用 ContinueWith 在 Windows 服务中重复任务 (TPL)

Question

我有一个 Windows 服务（用 C# 编写），它使用任务并行库 dll 来执行一些并行任务（一次 5 个任务） 在任务执行一次后，我想持续（每小时）重复相同的任务。调用 QueuePeek 方法 我是否像在下面的代码 sn-p 中设置的那样使用计时器或计数器？

我正在使用计数器来设置任务，一旦达到 5 我退出循环，但我也使用 .ContinueWith 来减少计数器，所以我的想法是计数器值会低于 5 因此循环会继续。但是我的 ContinueWith 似乎正在主线程上执行，然后循环退出。 使用 ContinueWith 调用 DecrementCounter 似乎不起作用

仅供参考：Importer 类是使用 MEF 加载一些库并完成工作

这是我的代码示例：

    private void QueuePeek()
    {            
        var list = SetUpJobs();

        while (taskCounter < 5)
        {
            int j = taskCounter;

            Task task = null;
            task = new Task(() =>
            {
                DoLoad(j);                                        
            });               

            taskCounter += 1;
            tasks[j] = task;
            task.ContinueWith((t) => DecrementTaskCounter());
            task.Start();
            ds.SetJobStatus(1);                               
        }            
        if (taskCounter == 0)
           Console.WriteLine("Completed all tasks.");
    }       

    private void DoLoad(int i)
    {
        ILoader loader;

    DataService.DataService ds = new DataService.DataService();

    Dictionary<int, dynamic> results = ds.AssignRequest(i);

    var data = results.Where(x => x.Key == 2).First();
    int loaderId = (int)data.Value;

    Importer imp = new Importer();

    loader = imp.Run(GetLoaderType(loaderId));

    LoaderProcessor lp = new LoaderProcessor(loader);

    lp.ExecuteLoader();
    }

    private void DecrementTaskCounter()
    {
        Console.WriteLine(string.Format("Decrementing task counter with threadId: {0}",Thread.CurrentThread.ManagedThreadId) );
        taskCounter--;
    }

Answer 1

我发现您的代码存在一些问题，这些问题可能会导致一些难以追踪的错误。首先，如果使用所有任务可能同时读取和写入的计数器，请尝试使用 Interlocked。例如：

Interlocked.Increment(ref _taskCounter); // or Interlocked.Decrement(ref _taskCounter);

如果我了解您要完成的工作，我认为您想要做的是使用一个计时器，您可以在每组任务完成后重新安排该计时器。

public class Worker
{
    private System.Threading.Timer _timer;
    private int _timeUntilNextCall = 3600000;

    public void Start()
    {
        _timer = new Timer(new TimerCallback(QueuePeek), null, 0, Timeout.Infinite);
    }
    private void QueuePeek(object state)
    {
        int numberOfTasks = 5;

        Task[] tasks = new Task[numberOfTasks];
        for(int i = 0; i < numberOfTasks; i++)
        {
            tasks[i] = new Task(() =>
            {
                DoLoad();
            });
            tasks[i].Start();
        }

        // When all tasks are complete, set to run this method again in x milliseconds
        Task.Factory.ContinueWhenAll(tasks, (t) => { _timer.Change(_timeUntilNextCall, Timeout.Infinite); });
    }

    private void DoLoad() { }
}