Поток данных TPL - контроль того, какие элементы находятся в потоке в любой момент

Мне нужно обработать файлы, которые нам присылают продавцы. Каждое имя файла будет содержать идентификатор продавца. Например. Товары - 118763.csv, где число - идентификатор продавца.

Я хочу настроить поток данных для обработки этих файлов. Одновременно можно обрабатывать несколько файлов.

Однако файлы для каждого продавца должны обрабатываться по порядку. Это означает, что никакие два файла для конкретного продавца не могут обрабатываться одновременно.

Я не уверен, как смоделировать это с помощью потока данных. Я мог бы отслеживать, какие продавцы обрабатываются в настоящее время, но я не знаю, как убедиться, что я не обрабатываю два файла от одного и того же продавца, при этом обрабатывая как можно больше файлов одновременно.

Кажется, мне нужно отфильтровать то, что я беру из очереди, но я могу получать только один элемент за раз, а когда у меня есть, что тогда. Я мог бы повторно запросить его, но это может означать много повторного запроса, если есть только два файла, и они оба от одного и того же продавца.

Любые идеи?


.net queue c# task-parallel-library tpl-dataflow
person bornfromanegg    schedule 18.07.2014    source источник
comment
возможный дубликат хэшированных / сегментированных блоков действий   -  person i3arnon    schedule 19.07.2014

Ответы (1)


arrow_upward
2
arrow_downward

Вы можете создать несколько ActionBlock (столько, сколько вы хотите обрабатывать параллельно), где каждый обрабатывает только один файл за раз, и используйте номер в имени файла, чтобы выбрать правильный ActionBlock. Это позволит параллельную обработку, в то время как файлы с одним и тем же идентификатором продавца обрабатываются последовательно.

Инициализация:

_actionBlocks = new ActionBlock<File>[DegreeOfParallelism];
for (var i = 0; i < _actionBlocks.Length; i++)
{
    _actionBlocks[i] = new ActionBlock<File>(file => ProcessFile(file));
}

Использование:

void ProcessFile(string path)
{
    _actionBlocks[(uint) int.Parse(Path.GetFileNameWithoutExtension(path))%_actionBlocks.Length].Post(
        ReadFile(path));
}
person i3arnon    schedule 19.07.2014