1

有一堆项目不断添加到数据库中，需要处理。我希望并行处理不相似的项目。

例如：

A 类物品：第 1 项、第 2 项、第 3 项

B 类项目：第 4 项、第 5 项、第 6 项

C 类项目：第 7 项、第 8 项、第 9 项

第 1 项、第 4 项和第 7 项应并行处理。

随着越来越多的某个类型的项目被添加到数据库中，它们将被选中并排队等待处理，只有在处理完该类型的先前项目之后。

我想我可以使用带有 CustomTaskScheduler 的静态任务工厂来做到这一点，它只会在该类型的上一个任务完成后才开始一个新任务？我的问题是我的 CustomTaskScheduler 应该如何看？

class test
{
    private static void Main()
    {
        //List of items from the database
        var itemList = new List<Item>();
        itemList.Add(new Item(1, "A"));
        itemList.Add(new Item(2, "A"));
        itemList.Add(new Item(3, "A"));

        itemList.Add(new Item(4, "B"));
        itemList.Add(new Item(5, "B"));
        itemList.Add(new Item(6, "B"));

        itemList.Add(new Item(7, "C"));
        itemList.Add(new Item(8, "C"));
        itemList.Add(new Item(9, "C"));

        //This needs to be run on a timer picking up new items from the database every time
        new ProcessQueue().ProcessAllItems(itemList);

        Console.ReadLine();
    }
}

public class ProcessQueue
{
    private static CustomTaskScheduler customTaskScheduler = new CustomTaskScheduler(1);
    private static TaskFactory factory = new TaskFactory(customTaskScheduler);

    public void ProcessAllItems(List<Item> itemList)
    {
        var cts = new CancellationTokenSource();

        foreach (var item in itemList)
        {
            factory.StartNew(
                o =>
                executeTask(item.Id, item.ItemType),
                item.ItemType, //unique identifier for multiple threads
                cts.Token);
        }

    }

    public void executeTask(int id, string parentId)
    {
        Console.WriteLine("Item - {0} ItemType - {1} on thread {1}   ", id, parentId,
                          Thread.CurrentThread.ManagedThreadId);
        Thread.Sleep(5000);
    }
}

public class Item
{
    public Item(int id, string itemType)
    {
        Id = id;
        ItemType = itemType;
    }

    public int Id { get; set; }
    public string ItemType { get; set; }
}
4

1 回答 1

0

这个怎么样，改编自How to: Create a Task Scheduler That Limits Concurrency。它使用 Task.AsyncState 来同步线程，它应该是你的 Item.ItemType 给你的例子。

public class ParallelTypeScheduler : TaskScheduler {

    // Indicates whether the current thread is processing work items.
   [ThreadStatic]
   private static bool _currentThreadIsProcessingItems;

   // The list of tasks to be executed  
   private readonly ConcurrentDictionary<object, LinkedList<Task>> _tasks = new ConcurrentDictionary<object, LinkedList<Task>>; // protected by lock(_tasks) 

   // Indicates whether the scheduler is currently processing work items.  
   private readonly Dictionary<object, bool> _typesRunning = new Dictionary<object,bool>();

   // Queues a task to the scheduler.  
   protected sealed override void QueueTask(Task task)
   {
        LinkedList<Task> typesTasks = _tasks.GetOrAdd(task.AsyncState, new LinkedList<Task>());

        lock (typesTasks)
        {
            typesTasks.AddLast(task);

            if(!_typesRunning.ContainsKey(task.AsyncState) || _typesRunning[task.AsyncState] == false){
                _typesRunning[task.AsyncState] = true;
                NotifyThreadPoolOfPendingWork(task.AsyncState);
            }
        }
   }

   // Inform the ThreadPool that there's work to be executed for this scheduler.  
   private void NotifyThreadPoolOfPendingWork(object type)
   {
       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
           {
               LinkedList<Task> typedTasks;
               if (_tasks.TryGetValue(type, out typedTasks)) {

                   while (true) {

                       Task item;
                       lock (typedTasks)
                       {
                           // When there are no more items to be processed, 
                           // note that we're done processing, and get out. 
                           if (_tasks.Count == 0)
                           {
                               _typesRunning[type] = false;
                               break;
                           }

                           // Get the next item from the queue
                           item = typedTasks.First.Value;
                           typedTasks.RemoveFirst();
                       }

                       base.TryExecuteTask(item);

                   }

               }
           }
           // 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) {

          LinkedList<Task> typedTasks;
          // Try to run the task.  
          if (_tasks.TryGetValue(task.AsyncState, out typedTasks) && TryDequeue(typedTasks, 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(LinkedList<Task> typedTasks, Task task)
   {
       lock (typedTasks) return typedTasks.Remove(task);
   }

   // 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.SelectMany(t => t.Value);
           else throw new NotSupportedException();
       }
       finally
       {
           if (lockTaken) Monitor.Exit(_tasks);
       }
   }

}

我不确定线程​​安全如何与 GetScheduledTasks 一起工作。

于 2014-11-27T00:33:10.247 回答