循环内的任务

Question

我有一个 Windows 服务，其线程每 2 分钟运行一次。

while (true)
{
    try
    {
        repNeg.willExecuteLoopWithTasks(param1, param2, param3);
        Thread.Sleep(20000);
    }

在这里面我有一个任务循环：

foreach (RepModel repModelo in listaRep)
{
    Task t = new Task(() => { this.coletaFunc(repModelo.EndIp, user, tipoFilial); });
    t.Start();
}

但我认为这个实现是错误的。我只需要为列表中的每个元素运行一个任务， 并且，当特定任务完成时，请稍等片刻，然后重新开始。

M8 我得说我这里有两种情况。

1 - 我等不及所有任务完成。因为有些任务可能需要 2 个小时以上才能完成，而另一些任务可能只需要 27 秒。

2 - 我的任务列表可以更改。这就是为什么我有一个线程。每 2 分钟我的线程获取一个要执行的任务列表，然后开始一个循环。

但有时我的任务还没有完成，另一个线程重新开始，然后我的日志中出现奇怪的事情。 我尝试使用字典来解决我的问题，但经过一段时间的执行，有时需要几天，我的日志显示：

“System.IndexOutOfRangeException”

Answer 1

这就是我要做的......

创建一个存储以下内容（作为属性）的新类：

• RepModel ID（唯一的）
• DateTime 最后一次运行
• int 表示任务应运行的频率（以秒为单位）
• bool 以确定任务是否正在进行中

然后你需要一个类的全局列表，比如说“JobList”。

您的主应用程序应该有一个计时器，它每隔几分钟运行一次。这个计时器的工作是检查新的RepModel（假设这些可以随着时间而改变，即数据库列表）。当这个勾选时，循环列表并将任何新的（不同的 ID）添加到JobList。您可能还想删除任何不再需要的内容（即从数据库列表中删除）。

然后你有第二个计时器，它每秒运行一次。它的工作是检查JobList 中的所有项目并将上次运行时间与当前时间进行比较（并确保它们尚未在进行中）。如果持续时间已经过去，则开始任务。任务完成后，更新上次运行时间，以便下次运行，确保随时更改“进行中”标志。

这都是理论，您需要自己尝试一下，但我认为它涵盖了您实际想要实现的目标。

---

一些示例代码（可能编译/工作）：

class Job
{
    public int ID { get; set; }
    public DateTime? LastRun { get; set; }
    public int Frequency { get; set; }
    public bool InProgress { get; set; }
}

List<Job> JobList = new List<Job>();

// Every 2 minutes (or whatever).
void timerMain_Tick()
{
    foreach (RepModel repModelo in listaRep)
    {
        if(!JobList.Any(x => x.ID == repModelo.ID)
        {
            JobList.Add(new Job(){ ID = repModel.ID, Frequency = 120 });
        }
    }
}

// Every 10 seconds (or whatever).
void timerTask_Tick()
{
    foreach(var job in JobList.Where(x => !x.InProgress && (x.LastRun == null || DateTime.Compare(x.LastRun.AddSeconds(x.Duration), DateTime.Now) < 0))
    {
        Task t = new Task(() => { 
            // Do task.
        }).ContinueWith(task => {
            job.LastRun = DateTime.Now;
            job.InProgress = false;
        }, TaskScheduler.FromCurrentSynchronizationContext());;
        job.InProgress = true;
        t.Start();
    }
}

Answer 2

所以你真正需要的是一个有两个操作的类，它需要能够开始处理你的一个模型，它需要能够结束你的一个模型的处理。将其从列表中分离出来会更容易。

当您开始处理模型时，您需要创建一个 CancellationTokenSource 以与其关联，以便您以后可以停止处理它。在您的情况下，处理它意味着有一个循环，虽然没有取消，但它运行一个操作然后等待一段时间。结束操作就像取消令牌源一样简单。

public class Foo
{
    private ConcurrentDictionary<RepModel, CancellationTokenSource> tokenLookup =
        new ConcurrentDictionary<RepModel, CancellationTokenSource>();
    public async Task Start(RepModel model)
    {
        var cts = new CancellationTokenSource();
        tokenLookup[model] = cts;
        while (!cts.IsCancellationRequested)
        {
            await Task.Run(() => model.DoWork());
            await Task.Delay(TimeSpan.FromMinutes(1));
        }
    }

    public void End(RepModel model)
    {
        CancellationTokenSource cts;
        if (tokenLookup.TryRemove(model, out cts))
            cts.Cancel();
    }
}

Answer 3

如果您正在使用框架4.0 等，您可以尝试从中受益

Parallel.ForEach

执行一个 foreach 操作，其中迭代可以并行运行。

并行代码可能如下所示：

Parallel.ForEach(listaRep , repModelo => {
     this.coletaFunc(repModelo.EndIp, user, tipoFilial);

});

这将在多个内核上运行（如果可能的话），并且您不需要某些特定的任务调度器，因为您的代码将等待并行循环内的所有并行任务完成。如果满足条件，您可以递归调用相同的函数。