9

我想实施一个优先的ActionBlock<T>. 这样我就可以有条件地TInput使用Predicate<T>.
我阅读了 Parallel Extensions Extras Samples和Guide to Implementing Custom TPL Dataflow Blocks。
但是仍然不知道我该如何实现这种情况。
- - - - - - - - - - - - - - 编辑 - - - - - - - - - - - ------
有一些任务，其中5个可以同时运行。当用户按下按钮时，一些（取决于谓词函数）任务应该以最高优先级运行。
事实上我写了这段代码

TaskScheduler taskSchedulerHighPriority;
ActionBlock<CustomObject> actionBlockLow;
ActionBlock<CustomObject> actionBlockHigh;
...
queuedTaskScheduler = new QueuedTaskScheduler(TaskScheduler.Default, 5);
taskSchedulerHigh = queuedTaskScheduler.ActivateNewQueue(0);
taskSchedulerLow = queuedTaskScheduler.ActivateNewQueue(1);
...
actionBlockHigh = new ActionBlock<CustomObject>(new Action<CustomObject>(method), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, SingleProducerConstrained = false, TaskScheduler = taskSchedulerHigh });
actionBlockLow = new ActionBlock<CustomObject>(new Action<CustomObject>(method), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, MaxMessagesPerTask = 1, TaskScheduler = taskSchedulerLow });
...     
if (predicate(customObject))
    actionBlockHigh.Post(customObject);
else
    actionBlockLow.Post(customObject);

但似乎优先级根本没有生效。
---------------------------------------- 编辑 ------------------
我找到了事实上，当我使用这行代码时：

actionBlockHigh = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { TaskScheduler = taskSchedulerHigh });
actionBlockLow = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { TaskScheduler = taskSchedulerLow });

导致应用程序正确观察Tasks的优先级，但一次只能执行一个任务，同时使用流程中显示的第一个代码块，导致应用程序同时运行5个任务但优先级顺序不合适。

actionBlockHigh = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, TaskScheduler = taskSchedulerHigh });
actionBlockLow = new ActionBlock<AvlHistory>(new Action<AvlHistory>(SemaphoreAction), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, TaskScheduler = taskSchedulerLow });

更新：
要 svick 的坦克，我应该MaxMessagesPerTask指定taskSchedulerLow.

4

1 回答 1

8

您的问题不包含很多细节，因此以下只是您可能需要的猜测。

我认为最简单的方法是让两个s 在ParallelExtensionsExtrasActionBlock上以不同的优先级运行。您将使用谓词链接到高优先级的链接，然后链接到低优先级的链接。此外，为了确保高优先级的 s 不等待，设置低优先级的块。QueuedTaskSchedulerTaskMaxMessagesPerTask

在代码中，它看起来像：

static ITargetBlock<T> CreatePrioritizedActionBlock<T>(
    Action<T> action, Predicate<T> isPrioritizedPredicate)
{
    var buffer = new BufferBlock<T>();

    var scheduler = new QueuedTaskScheduler(1);

    var highPriorityScheduler = scheduler.ActivateNewQueue(0);
    var lowPriorityScheduler = scheduler.ActivateNewQueue(1);

    var highPriorityBlock = new ActionBlock<T>(
        action, new ExecutionDataflowBlockOptions
        {
            TaskScheduler = highPriorityScheduler
        });
    var lowPriorityBlock = new ActionBlock<T>(
        action, new ExecutionDataflowBlockOptions
        {
            TaskScheduler = lowPriorityScheduler,
            MaxMessagesPerTask = 1
        });

    buffer.LinkTo(highPriorityBlock, isPrioritizedPredicate);
    buffer.LinkTo(lowPriorityBlock);

    return buffer;
}

这只是您可以做什么的草图，例如，Completion返回的块的行为不正确。

于 2012-12-14T22:17:05.670 回答