-2

嗨，我是一名实习生，几乎没有 C# 经验，他陷入了接管使用TaskCompletionSource和BlockingCollection实现多线程的 Windows 服务的境地。我从来没有做过 C#。我正在尝试优化服务处理其任务的方式，即处理日志文件。

我的问题是，BlockingCollection用于创建执行的线程队列WorkiItem，如何获取队列中活动线程的计数？这意味着有多少项已被EnqueueTask()命令调用仍处于运行状态？我不想要_taskQ.Count返回的队列积压计数。我想要活动线程的计数。我想将线程数保持在四个，并且只有在前一个项目完成后才将一个项目入队。我不想要队列中的项目黑日志。

public class ProducerConsumerQueue
{
    public CancellationTokenSource Token { get; set; }
    private BlockingCollection<WorkItem> _taskQ;

    public ProducerConsumerQueue(int workerCount)
    {
        _taskQ = new BlockingCollection<WorkItem>();
        for(int i = 0; i <workerCount; i++)
        {
            Task.Factory.StartNew(Consume);
        }
    }

    
    public Task EnqueueTask(Action action, CancellationToken? cancelToken)
    {
        var tcs = new TaskCompletionSource<object>();
        _taskQ.Add(new WorkItem(tcs, action, cancelToken));
        return tcs.Task;
    }

    public void Consume()
    {
        foreach (WorkItem workItem in _taskQ.GetConsumingEnumerable())
        {
            if (workItem.CancelToken.HasValue &&
                workItem.CancelToken.Value.IsCancellationRequested)
            {
                workItem.TaskSource.SetCanceled();
            }
            else
            {
                try
                {
                    workItem.Action();
                    workItem.TaskSource.SetResult(null);
                }
                catch (OperationCanceledException ex)
                {
                    if (ex.CancellationToken == workItem.CancelToken)
                    {
                        workItem.TaskSource.SetCanceled();
                    }
                    else
                    {
                        workItem.TaskSource.SetException(ex);
                    }
                }
                catch (Exception ex)
                {
                    workItem.TaskSource.SetException(ex);
                }
            }
        }
    }
}

这个ProducerCOnsumer队列在服务启动时被调用，它的队列在每个服务轮询间隔被重新加载。我想通过将其设置为线程安全数据库表中文件的限制来限制这些线程中产生的数量。因此，如果线程数为 4，则 db 表中的文件数将为 4。在完成 1 个文件之前，队列不应产生额外的线程或将文件排入队列。为此，我想一个简单的解决方案是计算活动线程的数量（这意味着文件处理的活动数量并且在线程下降 1 之前不要添加任何新文件：

protected override void OnStart(string[] args)
{
    ProducerConsumerQueue = new ProducerConsumerQueue(Constants.THREAD_COUNT);
    InitializeLogging();
    PollOnServiceStart();

    _timer.Elapsed += OnElapsedTime;
    _timer.Enabled = true;
    _timer.Interval = _interval;
}

public void OnElapsedTime(object source, ElapsedEventArgs args)
{
    try
    {
        //InitializeLogging();
        Poll();
    }
    catch (Exception ex)
    {
        Logger.Error(ex.Message.ToString());
    }
}


public void PollOnServiceStart()
{
    foreach (var handler in handlers)
    {
        ProducerConsumerQueue.EnqueueTask(handler.Execute, CancellationTokenSource.Token);
    }
}
4

1 回答 1

2

如果你想要一个线程安全的计数器来检查活动任务的数量，你可以使用一个int _counter字段和Interlocked.Increment(ref _counter)/ Interlocked.Decrement(ref _counter)。

try请记住在块中递减块中的第一行时递增，finally以便在引发异常时不会丢失任何调用。

https://docs.microsoft.com/en-us/dotnet/api/system.threading.interlocked.increment?view=net-5.0

https://docs.microsoft.com/en-us/dotnet/api/system.threading.interlocked.decrement?view=net-5.0

于 2021-02-05T21:57:28.530 回答