0

我有一个可以由多个线程（ConcurrentQueue<MyJob>）填充的作业队列。我需要异步（不是通过主线程）实现这些作业的连续执行，但同时只能由一个线程执行。我试过这样的事情：

public class ConcurrentLoop {
    private static ConcurrentQueue<MyJob> _concurrentQueue = new ConcurrentQueue<MyJob>();

    private static Task _currentTask;
    private static object _lock = new object();

    public static void QueueJob(Job job)
    {
        _concurrentQueue.Enqueue(job);
        checkLoop();
    }

    private static void checkLoop()
    {
        if ( _currentTask == null || _currentTask.IsCompleted )
        {
            lock (_lock)
            {
                if ( _currentTask == null || _currentTask.IsCompleted )
                {
                    _currentTask = Task.Run(() =>
                    {
                            MyJob current;
                            while( _concurrentQueue.TryDequeue( out current ) ) 
                                //Do something                                                       
                    });
                }
            }
        }
    }
}

我认为这段代码有一个问题：如果要执行的任务完成（TryDequeue返回false，但任务尚未标记为已完成）并且此时我得到了一份新工作，它将不会被执行。我对吗？如果是这样，如何解决这个问题

4

2 回答 2

3

您的问题陈述看起来像一个生产者-消费者问题，但需要注意的是您只需要一个消费者。

无需手动重新实现此类功能。相反，我建议使用BlockingCollection-- 它在内部使用ConcurrentQueue和一个单独的线程用于消费。请注意，这可能适合也可能不适合您的用例。

就像是：

_blockingCollection = new BlockingCollection<your type>(); // you may want to create bounded or unbounded collection
_consumingThread = new Thread(() =>
{
    foreach (var workItem in _blockingCollection.GetConsumingEnumerable()) // blocks when there is no more work to do, continues whenever a new item is added.
    {
      // do work with workItem
     }
});
_consumingThread.Start();

多个生产者（任务或线程）可以添加工作项_blockingCollection没有问题，并且无需担心同步生产者/消费者。

完成生产任务后，调用_blockingCollection.CompleteAdding()（此方法不是线程安全的，因此建议提前停止所有生产者）。可能，您还应该做一些事情_consumingThread.Join()来终止您的消费线程。

于 2016-09-19T22:50:49.450 回答
0

为此，我会使用 Microsoft 的 Reactive Framework Team 的 Reactive Extensions (NuGet "System.Reactive")。这是一个可爱的抽象。

public class ConcurrentLoop
{
    private static Subject<MyJob> _jobs = new Subject<MyJob>();

    private static IDisposable _subscription =
        _jobs
            .Synchronize()
            .ObserveOn(Scheduler.Default)
            .Subscribe(job =>
            {
                //Do something
            });

    public static void QueueJob(MyJob job)
    {
        _jobs.OnNext(job);
    }
}

这很好地将所有传入的作业同步到一个流中并将执行推送到Scheduler.Default（基本上是线程池），但是因为它已经序列化了所有输入，一次只能发生一个。这样做的好处是，如果值之间存在显着差距，它会释放线程。这是一个非常精简的解决方案。

要清理，您只需要调用_jobs.OnCompleted();或_subscription.Dispose();。

于 2016-09-20T00:05:25.230 回答