使用阻塞收集创建文件拾取进程

static Timer _aTimer = new System.Timers.Timer();
    static void Main(string[] args)
    {
        _aTimer.Elapsed += new ElapsedEventHandler(OnTimedEvent);
        _aTimer.Interval = 5000;
        _aTimer.Enabled = true;
        Console.WriteLine("Press ''q'' to quit the sample.");
        while (Console.Read() != 'q') ;
    }

        private static void OnTimedEvent(object source, ElapsedEventArgs e)
    {
        // stop the timer so we dont reprocess files we already have in the queue
        StopTimer();
        // setup a list of queues
        var lists = new List<IncomingOrderQueue>();
        //get the accounts in which the files we are looking in
        var accounts = new List<string>() { "Account1", "Account2" };
        //loop through the accounts and set up the queue 
        foreach (var acc in accounts)
        {
            // create the queue
            var tmp = new IncomingOrderQueue();
            // for each file in the folders add it to be processed in the queue
            foreach (var orderFile in OrderFiles(acc))
            {
                tmp.EnqueueSweep(new QueueVariables() { Account = acc, File = orderFile });
            }
            // add the queue to the list of queues
            lists.Add(tmp);
        }
        // for each of the queues consume all the contents of them
        Parallel.ForEach(lists, l => l.Consume());
        //start the timer back up again because we have finished all the files we have in the current queue
        StartTimer();
    }
        public static void StopTimer()
    {
        Console.WriteLine("Stop Timer");
        _aTimer.Stop();
        _aTimer.Enabled = false;
    }
    public static void StartTimer()
    {
        Console.WriteLine("Start Timer");
        _aTimer.Enabled = true;
        _aTimer.Start();
    }

 public class IncomingOrderQueue 
{
    BlockingCollection<QueueVariables> _orderQ = new BlockingCollection<QueueVariables>();
    public void EnqueueSweep(QueueVariables incoming)
    {
        // add items to the queue
        _orderQ.Add(incoming);
    }
    public void Consume()
    {
        // stop anything been adding to the queue
        _orderQ.CompleteAdding();
        // consume all the objects in the blocking collection
        Parallel.ForEach(_orderQ.GetConsumingEnumerable(), Processor.Order.Object);
    }
    public int QueueCount
    {
        get
        {
            return _orderQ.Count;
        }
    }
}

我认为你不需要停止和启动你的生产者和消费者。如果BlockingCollection达到最大容量，则可以阻止生产者，如果它是空的，则可以阻止消费者。

我也可能从单个BlockingCollection开始，直到分析显示我需要另一个。根据你的生产者和消费者的相对速度，你可能需要调整他们的数量。如果它们是IO绑定的，它们应该是异步的，你可以有很多，如果它们是CPU绑定的，你可能不需要比可用的处理器数量更多的处理器。

我已经重新做了你的例子假设IO绑定生产者和消费者，希望它给你一些想法。它以10秒的间隔发射生产者，并可以继续进行，直到您通过CanellationToken取消生产。只有在您取消并完成生产之后，您才能CompleteAdding释放被阻塞的消费者。

public class QueueVariables
{
    public string Account {get;set;}
    public string File {get;set;}
}
public static ConcurrentQueue<string> GetACcounts()
{
    return new ConcurrentQueue<string>(new []
        {
        "Account1",
        "Account2",
        "Account3",
        "Account4",
        "Account5",
        "Account6",
        "Account7",
        "Account8",
        "Account9",
        "Account10",
        "Account11",
        "Account12",
    });
}
public static List<string> GetFiles(string acct)
{
    return new List<string>
    {
        "File1",
        "File2",
        "File3",
        "File4",
        "File5",
        "File6",
        "File7",
        "File8",
        "File9",
        "File10",
        "File11",
        "File12",
    };
}
public static async Task StartPeriodicProducers(int numProducers, TimeSpan period, CancellationToken ct)
{
    while(!ct.IsCancellationRequested)
    {
        var producers = StartProducers(numProducers, ct);
        // wait for production to finish
        await Task.WhenAll(producers.ToArray());
        // wait before running again
        Console.WriteLine("***Waiting " + period);
        await Task.Delay(period, ct);
    }
}
public static List<Task> StartProducers(int numProducers, CancellationToken ct)
{
    List<Task> producingTasks = new List<Task>();
    var accounts = GetACcounts();
    for (int i = 0; i < numProducers; i++)
    {
        producingTasks.Add(Task.Run(async () =>
        {
            string acct;
            while(accounts.TryDequeue(out acct) && !ct.IsCancellationRequested)
            {
                foreach (var file in GetFiles(acct))
                {
                    _orderQ.Add(new UserQuery.QueueVariables{ Account = acct, File = file });
                    Console.WriteLine("Produced Account:{0} File:{1}", acct, file);
                    await Task.Delay(50, ct); // simulate production delay
                }
            }
            Console.WriteLine("Finished producing");
        }));
    }
    return producingTasks;
}
public static List<Task> StartConsumers(int numConsumers)
{
    List<Task> consumingTasks = new List<Task>();
    for (int j = 0; j < numConsumers; j++)
    {
        consumingTasks.Add(Task.Run(async () =>
        {
            try
            {
                while(true)
                {
                    var queueVar = _orderQ.Take();
                    Console.WriteLine("Consumed Account:{0} File:{1}", queueVar.Account, queueVar.File);
                    await Task.Delay(200); // simulate consumption delay
                }
            }
            catch(InvalidOperationException)
            {
                Console.WriteLine("Finished Consuming");
            }
        }));
    }
    return consumingTasks;
}
private static async Task MainAsync()
{
    CancellationTokenSource cts = new CancellationTokenSource();
    var periodicProducers = StartPeriodicProducers(2, TimeSpan.FromSeconds(10), cts.Token);
    var consumingTasks = StartConsumers(4);
    await Task.Delay(TimeSpan.FromSeconds(120));
    // stop production
    cts.Cancel();
    try
    {
        // wait for producers to finish producing
        await periodicProducers;
    }
    catch(OperationCanceledException)
    {
        // operation was cancelled
    }
    // complete adding to release blocked consumers
    _orderQ.CompleteAdding();
    // wait for consumers to finish consuming
    await Task.WhenAll(consumingTasks.ToArray());
}
// maximum size 10, after that capaicity is reached the producers block
private static BlockingCollection<QueueVariables> _orderQ = new BlockingCollection<QueueVariables>(10);
void Main()
{
    MainAsync().Wait();
    Console.ReadLine();
}
// Define other methods and classes here