【问题标题】:Task counter C#, Interlocked任务计数器 C#,互锁
【发布时间】:2018-02-06 10:02:45
【问题描述】:

我想并行运行任务,在给定时间运行的实例不超过 10 个。

这是我目前的代码:

private void Listen()
{
    while (true)
    {
        var context = listener.GetContext();
        var task = Task.Run(() => HandleContextAsync(context));
        Interlocked.Increment(ref countTask);
        if (countTask > 10)
        {
            //I save tasks in the collection
        }
        else
        {
            task.ContinueWith(delegate { Interlocked.Decrement(ref countTask); }); //I accomplish the task and reduce the counter
        }
    }
}

【问题讨论】:

  • 您的问题是:我的代码是否正确?如果是这样,那取决于你想要它做什么。
  • @pm_2 是的,我的问题是:我的代码是否正确?我想限制使用任务计数器的任务数量。一旦任务数量达到限制,我将它们放入队列中。所以我想用任务计数器和队列来实现节流。
  • 所以你想启动定义数量的任务(在本例中为 10 个)?
  • @pm_2 是的。
  • @mjwills countTask 是一个线程安全计数器,用于限制任务数

标签: c# task


【解决方案1】:

我建议您使用并行循环;例如:

Parallel.For(1, 10, a =>
{
      var context = listener.GetContext();
      ...
});

这将启动定义数量的任务,而无需您自己管理流程。

【讨论】:

    【解决方案2】:

    如果您想持续并行执行代码,一次最多 10 个实例,这可能值得考虑:

    private void Listen()
    {
        var options = new ParallelOptions() { MaxDegreeOfParallelism = 10 };
        Parallel.For(1, long.MaxValue - 1, options, (i) =>
        {
            var context = listener.GetContext();
            HandleContextAsync(context);
        });
    }
    

    基本上,它会不断地运行代码(大约long.MaxValue 次)。 MaxDegreeOfParallelism 确保它一次只运行 10 个代码“实例”。

    【讨论】:

      【解决方案3】:

      我假设 GetContext 的结果不是由您创建的,因此,当您不知道运行多少次或没有所有立即处理的上下文。

      因此,解决此问题的最佳方法可能是实现您自己的TaskScheduler。通过这种方式,您可以添加更多任务以按需解决,并具有固定的并发级别。

      基于Microsoft Docs 网站的示例,您已经可以实现这一点。

      我做了一个示例程序,对网站上的LimitedConcurrencyLevelTaskScheduler 进行了一些更改。

      using System;
      using System.Collections.Generic;
      using System.Threading;
      using System.Threading.Tasks;
      
      namespace parallel
      {
          class Program
          {
              private static Random Rand = new Random();
      
              static void Main(string[] args)
              {
                  var ts = new LimitedConcurrencyLevelTaskScheduler(10);
                  var taskFactory = new TaskFactory(ts);
      
                  while (true)
                  {
                      var context = GetContext(ts);
                      if (context.Equals("Q", StringComparison.OrdinalIgnoreCase))
                          break;
                      taskFactory.StartNew(() => HandleContextAsync(context));
                  }
                  Console.WriteLine("Waiting...");
      
                  while (ts.CountRunning != 0)
                  {
                      Console.WriteLine("Now running {0}x tasks with {1}x queued.", ts.CountRunning, ts.CountQueued);
                      Thread.Yield();
                      Thread.Sleep(100);
                  }
              }
      
              private static void HandleContextAsync(string context)
              {
                  // delays for 1-10 seconds to make the example easier to understand
                  Thread.Sleep(Rand.Next(1000, 10000));
      
                  Console.WriteLine("Context: {0}, from thread: {1}", context, Thread.CurrentThread.ManagedThreadId);
              }
      
              private static string GetContext(LimitedConcurrencyLevelTaskScheduler ts)
              {
                  Console.WriteLine("Now running {0}x tasks with {1}x queued.", ts.CountRunning, ts.CountQueued);
                  return Console.ReadLine();
              }
          }
          // Provides a task scheduler that ensures a maximum concurrency level while 
          // running on top of the thread pool.
          public class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
          {
              // Indicates whether the current thread is processing work items.
              [ThreadStatic]
              private static bool _currentThreadIsProcessingItems;
      
              // The list of tasks to be executed 
              private readonly LinkedList<Task> _tasks = new LinkedList<Task>(); // protected by lock(_tasks)
      
              public int CountRunning => _nowRunning;
      
              public int CountQueued
              {
                  get
                  {
                      lock (_tasks)
                      {
                          return _tasks.Count;
                      }
                  }
              }
      
              // The maximum concurrency level allowed by this scheduler. 
              private readonly int _maxDegreeOfParallelism;
      
              // Indicates whether the scheduler is currently processing work items. 
              private volatile int _delegatesQueuedOrRunning = 0;
              private volatile int _nowRunning;
      
              // Creates a new instance with the specified degree of parallelism. 
              public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
              {
                  if (maxDegreeOfParallelism < 1)
                      throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
                  _maxDegreeOfParallelism = maxDegreeOfParallelism;
              }
      
              // Queues a task to the scheduler. 
              protected sealed override void QueueTask(Task task)
              {
                  // Add the task to the list of tasks to be processed.  If there aren't enough 
                  // delegates currently queued or running to process tasks, schedule another. 
                  lock (_tasks)
                  {
                      _tasks.AddLast(task);
                      if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism)
                      {
                          Interlocked.Increment(ref _delegatesQueuedOrRunning);
                          NotifyThreadPoolOfPendingWork();
                      }
                  }
              }
      
              // Inform the ThreadPool that there's work to be executed for this scheduler. 
              private void NotifyThreadPoolOfPendingWork()
              {
                  ThreadPool.UnsafeQueueUserWorkItem(_ =>
                  {
                      // Note that the current thread is now processing work items.
                      // This is necessary to enable inlining of tasks into this thread.
                      _currentThreadIsProcessingItems = true;
                      try
                      {
                          // Process all available items in the queue.
                          while (true)
                          {
                              Task item;
                              lock (_tasks)
                              {
                                  // When there are no more items to be processed,
                                  // note that we're done processing, and get out.
                                  if (_tasks.Count == 0)
                                  {
                                      Interlocked.Decrement(ref _delegatesQueuedOrRunning);
                                      break;
                                  }
      
                                  // Get the next item from the queue
                                  item = _tasks.First.Value;
                                  _tasks.RemoveFirst();
                              }
      
                              // Execute the task we pulled out of the queue
                              Interlocked.Increment(ref _nowRunning);
                              if (base.TryExecuteTask(item))
                                  Interlocked.Decrement(ref _nowRunning);
                          }
                      }
                      // We're done processing items on the current thread
                      finally { _currentThreadIsProcessingItems = false; }
                  }, null);
              }
      
              // Attempts to execute the specified task on the current thread. 
              protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
              {
                  // If this thread isn't already processing a task, we don't support inlining
                  if (!_currentThreadIsProcessingItems) return false;
      
                  // If the task was previously queued, remove it from the queue
                  if (taskWasPreviouslyQueued)
                      // Try to run the task. 
                      if (TryDequeue(task))
                          return base.TryExecuteTask(task);
                      else
                          return false;
                  else
                      return base.TryExecuteTask(task);
              }
      
              // Attempt to remove a previously scheduled task from the scheduler. 
              protected sealed override bool TryDequeue(Task task)
              {
                  lock (_tasks) return _tasks.Remove(task);
              }
      
              // Gets the maximum concurrency level supported by this scheduler. 
              public sealed override int MaximumConcurrencyLevel { get { return _maxDegreeOfParallelism; } }
      
              // Gets an enumerable of the tasks currently scheduled on this scheduler. 
              protected sealed override IEnumerable<Task> GetScheduledTasks()
              {
                  bool lockTaken = false;
                  try
                  {
                      Monitor.TryEnter(_tasks, ref lockTaken);
                      if (lockTaken) return _tasks;
                      else throw new NotSupportedException();
                  }
                  finally
                  {
                      if (lockTaken) Monitor.Exit(_tasks);
                  }
              }
          }
      }
      

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 2022-07-31
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 2023-02-01
        • 1970-01-01
        • 2013-06-19
        相关资源
        最近更新 更多