Каталог решений - Многопоточная обработка данных на примере перепроведения документов

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

Общее описание механизма

Процесс начинается с запуска основного фонового задания (такой упрощенный менеджер потоков), в которое передаются все необходимые параметры. Далее формируется выборка объектов для обработки (документов). Выполняется обход выборки и выбирается первая порция документов (согласно переданному параметру «КоличествоДокументовВПорции»).

Затем происходит разбивка порции документов на независимые наборы, которые могут обрабатываться параллельно, не мешая друг другу. Именно этот момент самый интересный. 

Далее выполняется запуск необходимого количества потоков для обработки каждого набора документов (согласно переданному параметру «КоличествоПотоков»).

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

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

Самым интересным шагом в данной последовательности является деление выбранной порции на наборы не связанных друг с другом документов. Сразу оговорюсь, мой вариант не является окончательным. Он скорее показывает, как можно решить подобную задачу, и является плацдармом для дальнейшего развития механизма. На некоторых из тестируемых баз он позволяет создать достаточно ровную разбивку, на некоторых большая часть документов попадают в первый или во второй набор. Для реального проекта конечно следует использовать более сложную и «умную» разбивку.

Суть моей разбивки следующая. Не допускается попадание в разные наборы документов, которые создают движения по одним и тем же комбинациям измерений: номенклатура + характеристика + склад. Причем важны как приходные так и расходные документы, так как важно соблюсти последовательность проведения в рамках выше указанных измерений.

Сразу о минусах моей разбивки

1. Разбивка выполняется по данным регистра «Товары организаций», следовательно, она актуальна только для конфигураций из семейства ЕРП (ЕРП, КА, УТ).
2. В моем примере документы обязательно должны быть проведены, так как выборка формируется по данным регистра «Товары организаций» и используются элементы справочника «КлючиАналитикиУчетаНоменклатуры». Если исходные документы не проведены, выборку необходимо формировать другим способом.
3. Учитываются только возможное пересечение товарного состава. Можно учитывать еще и пересечения по партнерам и договорам для документов закупки и реализации.
4. Для определения зависимостей я использую элементы справочника «КлючиАналитикиУчетаНоменклатуры». В моем примере не ведется учет по сериям и обособленный учет товаров. Поэтому ключи аналитики как раз и являются объектами разделения по указанным выше измерениям. Если учет по сериям или обособленный учет присутствуют, ключи аналитики получаются с более детальной разбивкой, и опираться на них не совсем правильно. Регистр накопления «Свободные остатки», имеет среди измерений только: номенклатура, характеристика и склад. Если в разные наборы попадут документы с разными ключами аналитики, но с одинаковыми наборами измерений: номенклатура, характеристика и склад, будут возможны пересечения.

Реализация

Код модуля формы я приводить не будут, концептуально он не отличается от кода модуля формы из предыдущей статьи. В форме задается период и значения параметров: КоличествоДокументовВПорции, КоличествоПотоков. Также динамически создаются прогрессы выполнения отдельных потоков согласно параметру «КоличествоПотоков», для более удобного отслеживания выполнения. Прогрессы отображают состояния выполнения потоков и обрабатываемый документ в текущей порции. По состоянию общего (первого) прогресса можно судить об общем проценте выполнения задачи:

Процедуры модуля обработки

Не буду приводить код процедур: «СведенияОВнешнейОбработке»и «ВыполнитьКоманду». В них ничего особенного нет. Отмечу только, что относительно предыдущего варианта появилась дополнительная команда «ВыполнениеПотокаВФоне». Она необходимо для запуска отдельных потоков из основного фонового задания (менеджера потоков).

Точка входа основного фонового задания, это процедура «ВыполнитьПерепроведениеВПотоках»

Процедура ВыполнитьПерепроведениеВПотоках(ДатаНачала, ДатаОкончания, КоличествоПотоков, КоличествоДокументовВПорции)
	
    Запрос = новый Запрос("ВЫБРАТЬ
	|	КОЛИЧЕСТВО(РАЗЛИЧНЫЕ ТоварыОрганизаций.Регистратор) КАК ВсегоДокументов
	|ИЗ
	|	РегистрНакопления.ТоварыОрганизаций КАК ТоварыОрганизаций
	|ГДЕ
	|	ТоварыОрганизаций.Период >= &ДатаНачала
	|	И ТоварыОрганизаций.Период <= &ДатаОкончания
	|	И НЕ ТоварыОрганизаций.Регистратор ССЫЛКА Документ.ПередачаТоваровМеждуОрганизациями
	|;
	|
	|////////////////////////////////////////////////////////////////////////////////
	|ВЫБРАТЬ
	|	ТоварыОрганизаций.АналитикаУчетаНоменклатуры КАК Аналитика,
	|	ТоварыОрганизаций.Регистратор КАК Документ,
	|	ТоварыОрганизаций.Период КАК Период
	|ИЗ
	|	РегистрНакопления.ТоварыОрганизаций КАК ТоварыОрганизаций
	|ГДЕ
	|	ТоварыОрганизаций.Период >= &ДатаНачала
	|	И ТоварыОрганизаций.Период <= &ДатаОкончания
	|	И НЕ ТоварыОрганизаций.Регистратор ССЫЛКА Документ.ПередачаТоваровМеждуОрганизациями
	|
	|СГРУППИРОВАТЬ ПО
	|	ТоварыОрганизаций.АналитикаУчетаНоменклатуры,
	|	ТоварыОрганизаций.Регистратор,
	|	ТоварыОрганизаций.Период
	| УПОРЯДОЧИТЬ ПО
	|  Период,
	|  Документ");
	
    Запрос.УстановитьПараметр("ДатаНачала", НачалоДня(ДатаНачала));
    Запрос.УстановитьПараметр("ДатаОкончания", КонецДня(ДатаОкончания));

    Результат = Запрос.ВыполнитьПакет();
    Выборка = Результат[0].Выбрать();
    Выборка.Следующий();
    ВсегоДокументов = Выборка.ВсегоДокументов;
	
    ТЗСвязи = Результат[1].Выгрузить();

    //Новая таблица для данных текущей порции	
    ТЗСвязиВПорции = новый ТаблицаЗначений;
    ТЗСвязиВПорции.Колонки.Добавить("Аналитика", новый ОписаниеТипов("СправочникСсылка.КлючиАналитикиУчетаНоменклатуры"));
    ТЗСвязиВПорции.Колонки.Добавить("Документ");	
    ТЗСвязиВПорции.Колонки.Добавить("Период", новый ОписаниеТипов("Дата"));	
    ТЗСвязиВПорции.Колонки.Добавить("ЭтоПриход", новый ОписаниеТипов("ВидДвиженияНакопления"));	

    ДокументовВТекПорции = 0;
    ТекущийДокумент = Неопределено;
    НомерПорции = 0;

    Для каждого СтрокаСвязи Из ТЗСвязи Цикл
        
        Если ТекущийДокумент <> СтрокаСвязи.Документ Тогда
           
            //Если количество документов равно размеру порции, выполняем обработку данных
            Если ДокументовВТекПорции = КоличествоДокументовВПорции Тогда
                НомерПорции = НомерПорции + 1;
                ОбработатьПорциюДокументов(ТЗСвязиВПорции, КоличествоПотоков, ДокументовВТекПорции);
                ДлительныеОперации.СообщитьПрогресс(Формат(КоличествоДокументовВПорции*НомерПорции/ВсегоДокументов*100, "ЧЦ=3; ЧДЦ="), "Выполняется проведение документов, порция: "+Строка(НомерПорции+1));
                ТЗСвязиВПорции.Очистить();
                ДокументовВТекПорции = 0;
                ТекущийДокумент = Неопределено;
            КонецЕсли;
				
            ТекущийДокумент = СтрокаСвязи.Документ;
            ДокументовВТекПорции = ДокументовВТекПорции + 1;
        КонецЕсли;	
        
        НоваяСтрока = ТЗСвязиВПорции.Добавить();
        ЗаполнитьЗначенияСвойств(НоваяСтрока, СтрокаСвязи);

    КонецЦикла;

    //Обработка последней порции документов
    ОбработатьПорциюДокументов(ТЗСвязиВПорции, КоличествоПотоков, ДокументовВТекПорции);
	
КонецПроцедуры