Конвейер обработки задач

• Пакетная регистрация задач для обработки
• Настройка цепочки последовательных обработчиков (Обработчик1 → Обработчик2 → ОбработчикN)
• Многопоточная обработка задач
• Обработка ошибок и повторное выполнение задач
• Сохранение результатов промежуточного выполнения
• Параллельность выполнения различных конвейеров
• Гарантия последовательности обработки задач с одним ключом
• Фиксирование сообщений формируемых при выполнении задачи (например не просто ошибка проведения документа, но и покажет сообщения которые формировались в ходе проведения)
   

• Отложенное проведение документа с созданием подчиненных документов

• Надежная интеграция с внешней системой (отправка данных в шину)

• Многошаговый ETL-процесс (Последовательное выполнение запросов, трансформация данных, загрузка в хранилище)

• Обработка входящих файлов (массовое создание объектов)

// Создает и возвращает структуру с параметрами для регистрации задачи в очереди обработки.
//
// Возвращаемое значение:
//   Структура - со следующими свойствами:
//      * ИдентификаторЗадачи - УникальныйИдентификатор - уникальный идентификатор задачи.
//      * ПараметрыЗадачи - Произвольный - параметры для выполнения задачи.
//      * Ключ - Строка,ЛюбаяСсылка - ключ группировки задач.
//
Функция ПараметрыРегистрацииЗадачи() Экспорт

// Регистрирует пакет задач в системе для последующей обработки.
// Записывает задачи в регистр сведений пакетами указанного размера.
//
// Параметры:
//   ИдентификаторКонвейера - Строка - имя модуля-обработчика, в котором находится метод обработки.
//   МассивЗадач - Массив из см. ПараметрыРегистрацииЗадачи - массив задач для регистрации.
//   РазмерПорции - Число - [необязательный, по умолчанию 1] размер пакета для записи.
//   УдалитьПоКлючу - Булево - [необязательный, по умолчанию Ложь] признак необходимости предварительного удаления задач с таким же ключом.
//
Процедура РегистрацияЗадач(ИдентификаторКонвейера, МассивЗадач, РазмерПорции = 1, УдалитьПоКлючу = Ложь) Экспорт

// Регистрирует одну задачу в системе для последующей обработки.
// Обертка над процедурой РегистрацияЗадач.
//
// Параметры:
//   ИдентификаторКонвейера - Строка - имя модуля-обработчика.
//   Задача - см. ПараметрыРегистрацииЗадачи - задача для регистрации.
//   УдалитьПоКлючу - Булево - [необязательный, по умолчанию Ложь] признак необходимости предварительного удаления задач с таким же ключом.
//
Процедура РегистрацияЗадачи(ИдентификаторКонвейера, Задача, УдалитьПоКлючу = Ложь) Экспорт

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

// Запускает конвейер обработки задач в фоновом режиме.
// Создает и выполняет фоновое задание для обработки задач указанного модуля.
//
// Параметры:
//   Конвейер - СправочникСсылка.озКонвейерыОбработкиЗадач - конвейер для запуска.
//
// Возвращаемое значение:
//   УникальныйИдентификатор - идентификатор запущенного фонового задания или Неопределено, если конвейер уже выполняется.
//
Функция ЗапускКонвейера(Конвейер) Экспорт

Добавьте в конфигурацию объекты подсистемы 


Реализуйте обработчики в соответсвии с требованиям к сигнатуре (пример тестового модуля)

// Выполняет обработку задачи в конвейере. 
// Может использоваться для выполнения произвольной логики, например, обработки данных.
// В процессе выполнения может сгенерировать исключение (например, для тестирования обработки ошибок).
//
// Параметры:
//   ПараметрыЗадачи - Произвольный - входные параметры задачи. Могут быть любого типа, переданы из предыдущего обработчика или при регистрации.
//   Кэш - Соответствие - используется для кэширования данных между задачами в рамках одного пакета обработки. 
//          Например, можно хранить соединение с базой данных, чтобы не открывать его для каждой задачи.
//   ЭтоПоследняя - Булево - признак того, что это последняя задача в текущем пакете. 
//          Можно использовать для освобождения ресурсов (например, закрытия соединения с базой).
//
Функция Обработчик1(ПараметрыЗадачи, Кэш, ЭтоПоследняя) Экспорт
	
	
	Сообщить("Делаем Обработчик1");
	
	Если ЗначениеЗаполнено(ПараметрыЗадачи)   Тогда
		ВызватьИсключение "Заполенены параметры....";
	КонецЕсли;
	
	СлучайнаяОшибка();
	
	Возврат "" + ПараметрыЗадачи + "/Обработчик1";
	
КонецФункции

Функция Обработчик2(ПараметрыЗадачи, Кэш, ЭтоПоследняя) Экспорт
	
	Сообщить("Делаем Обработчик2");
	 
	Если СтрНайти(ПараметрыЗадачи,"Обработчик1") = 0  Тогда
		ВызватьИсключение "не найден Обработчик1";
	КонецЕсли;
	
	СлучайнаяОшибка();
	
	Возврат ПараметрыЗадачи + "/Обработчик2";
	
КонецФункции

Функция Обработчик3(ПараметрыЗадачи, Кэш, ЭтоПоследняя) Экспорт
	
	Если СтрНайти(ПараметрыЗадачи,"Обработчик1") = 0  Тогда
		ВызватьИсключение "не найден Обработчик1";
	КонецЕсли;
	
	Если СтрНайти(ПараметрыЗадачи,"Обработчик2") = 0  Тогда
		ВызватьИсключение "не найден Обработчик2";
	КонецЕсли;
	
	Сообщить("Делаем Обработчик3");
	
	СлучайнаяОшибка();
	
	Возврат ПараметрыЗадачи + "/Обработчик3";
	
КонецФункции

Процедура СлучайнаяОшибка()
	Возврат;	  
	ГСЧ = Новый ГенераторСлучайныхЧисел; 
	Число = ГСЧ.СлучайноеЧисло(0, 100);
	
	Если Число%5 = 0 Тогда
		ВызватьИсключение "Случайная ошибка " + Число;
	КонецЕсли;
	
КонецПроцедуры      

Создайте элемент справочника "Конвейры обработки задач"

Идентификатор (код) - уникальный идентификатор который будете указывать при регистрации задач
Включен - признак влияет на запуск из рег. задания "(ОЗОН) Конвейры обработки задач"
Количество выбираемых задач - размер выборки для одного цикла запуска

Добавляете свои обработчики указывая:
Метод - полный путь к методу
Количество потоков - количество потоков для этого метода
Размер порции - максимальный размер порции (сообщения с одним ключом всегда попадают в один поток)
Максимальное время выполнения - в случае если фоновое задание потерялось и нет изменений по базе данных мы некоторое время будет проверять состояние задач пачки (нужно указывать максимальное время выполнение пачки)
Максимальное количество ошибок - в случае ошибки обработки задачи будут повторно выполняться при следующих запусках конвейра указанное количество раз 

Добавьте регистрацию задач в прикладной код

	
	МассивЗадач = Новый Массив;
	
	Для Сч = 1 По  КоличествоЗадач Цикл
		
		НоваяЗадача = озКонвейерОбработкиЗадач.ПараметрыРегистрацииЗадачи();
		НоваяЗадача.ПараметрыЗадачи = "";
		НоваяЗадача.Ключ = Строка(Сч); 
		МассивЗадач.Добавить(НоваяЗадача);
		
	КонецЦикла;
	
	озКонвейерОбработкиЗадач.РегистрацияЗадач("озТестМодульДляКонвейераОбработкиЗадач", МассивЗадач, 1000, Истина);

Настройте регламентное задание "Запуск конвейеров"
 

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

// Запускает конвейер обработки задач в фоновом режиме.
// Создает и выполняет фоновое задание для обработки задач указанного модуля.
//
// Параметры:
//   Конвейер - СправочникСсылка.озКонвейерыОбработкиЗадач - конвейер для запуска.
//
// Возвращаемое значение:
//   УникальныйИдентификатор - идентификатор запущенного фонового задания или Неопределено, если конвейер уже выполняется.
//
Функция ЗапускКонвейера(Конвейер) Экспорт