При разработке систем интеграции часто возникает ситуация, когда объем данных в регистре сведений становится настолько велик, что одно регламентное задание не успевает их обработать. Естественным решением кажется запуск нескольких параллельных потоков (регламентных заданий). Однако здесь мы сталкиваемся с серьезной проблемой: как сделать так, чтобы разные потоки не «подхватили» одни и те же записи и не отправили их в веб-сервис дважды?
Стандартный подход с блокировкой записи часто приводит к тому, что второе задание замирает в ожидании на 20 секунд (стандартный таймаут 1С), что сводит на нет всю эффективность параллелизма. В этой статье мы подробно разберем, как грамотно распределить нагрузку между заданиями, минимизировать ожидания и гарантировать уникальность обработки каждой записи.
Проанализируем причину, по которой автор темы столкнулся с задержкой в 20 секунд. По умолчанию 1С при попытке установить БлокировкаДанных, если ресурс уже занят другой транзакцией, ждет освобождения в течение времени, указанного в настройках сервера (обычно это 20 секунд). Если мы хотим, чтобы регламентное задание не ждало, а мгновенно переходило к следующей порции данных, нам нужно программно ограничить время ожидания.
Рассмотрим пример кода, который позволяет попытаться захватить запись и, если она занята, сразу продолжить работу:
Блокировка = Новый БлокировкаДанных;
ЭлементБлокировки = Блокировка.Добавить("РегистрСведений.ОчередьСобытий");
ЭлементБлокировки.УстановитьЗначение("ИдентификаторЗаписи", ТекущаяЗапись);
ЭлементБлокировки.Режим = РежимБлокировкиДанных.Исключительный;
// Устанавливаем время ожидания в 0 секунд
Блокировка.УстановитьВремяОжидания(0);
Попытка
Блокировка.Заблокировать();
// Если выполнение дошло сюда, значит запись наша, можем отправлять
ОтправитьДанныеНаСервер(ТекущаяЗапись);
Исключение
// Сюда попадем, если запись уже заблокирована другим потоком
// Просто переходим к следующей записи в цикле
Продолжить;
КонецПопытки;
Важный нюанс: Блокировка живет только до конца транзакции. Поэтому чтение данных, блокировка и сама отправка (или пометка о начале отправки) должны происходить в одном транзакционном блоке. Однако длительные сетевые вызовы внутри транзакции — это плохая практика. Поэтому лучше в транзакции только «столбить» запись (например, меняя ей статус), а саму отправку делать после фиксации.
Разберем методику, которая позволяет избежать создания лишних регистров, но требует добавления одного реквизита в существующий регистр. Мы добавим реквизит ИдентификаторПотока с типом УникальныйИдентификатор.
Логика работы будет следующей:
МойID = Новый УникальныйИдентификатор().ИдентификаторПотока пустой).МойID.Посмотрим на пример реализации захвата данных:
МойID = Новый УникальныйИдентификатор();
ПорцияДанных = 100;
НачатьТранзакцию();
// Выбираем первые N свободных записей
Запрос = Новый Запрос;
Запрос.Текст =
"ВЫБРАТЬ ПЕРВЫЕ " + ПорцияДанных + "
| Реестр.КлючЗаписи КАК Ключ
|ИЗ
| РегистрСведений.РеестрНаОтправку КАК Реестр
|ГДЕ
| Реестр.ИдентификаторПотока ЕСТЬ NULL
|ДЛЯ ИЗМЕНЕНИЯ"; // Используем для СУБД блокировки
Результат = Запрос.Выполнить().Выгрузить();
Для Каждого Строка Из Результат Цикл
МенеджерЗаписи = РегистрыСведений.РеестрНаОтправку.СоздатьМенеджерЗаписи();
// Заполняем отбор по ключу...
МенеджерЗаписи.Прочитать();
МенеджерЗаписи.ИдентификаторПотока = МойID;
МенеджерЗаписи.Записать();
КонецЦикла;
ЗафиксироватьТранзакцию();
// Теперь спокойно работаем с записями, где ИдентификаторПотока = МойID
Этот метод хорош тем, что он исключает конфликты блокировок на чтение. Каждое задание работает в своем «песочнице», определенной его GUID-ом.
Выясним, почему в сообщениях (1) и (9) советовали планы обмена. Механизм планов обмена в 1С изначально спроектирован для того, чтобы отслеживать изменения и гарантировать их доставку. Если мы регистрируем изменения в регистре по узлу, мы можем использовать метод ВыбратьИзменения().
При использовании планов обмена платформа сама управляет блокировками на уровне таблиц регистрации изменений. Когда одно регламентное задание вызывает ПланыОбмена.ВыбратьИзменения(Узел, НомерСообщения), эти данные помечаются этим номером сообщения, и другой поток (если вы используете разные узлы для потоков) их не увидит. Однако, если все РЗ работают по одному узлу, 1С обеспечит последовательный доступ к чтению изменений, что также предотвратит дублирование.
Проанализируем предложение из сообщения (7). Если мы не хотим менять структуру основного регистра (например, он типовой), можно создать вспомогательный регистр ЗаписиВРаботе.
Алгоритм работы:
КлючЗадания.ЗаписиВРаботе «зависших» данных от заданий, которые уже не активны?». Это можно сделать с помощью функции ФоновыеЗадания.ПолучитьФоновыеЗадания().Если задача не в том, чтобы обрабатывать данные параллельно, а в том, чтобы просто не мешать друг другу, достаточно использовать механизм проверки уникальности запуска. Разберем, как это сделать:
ПараметрыОтбора = Новый Структура;
ПараметрыОтбора.Вставить("ИмяМетаданных", "РегламентноеЗадание.ОтправкаДанныхВСервис");
ПараметрыОтбора.Вставить("Состояние", СостояниеФоновогоЗадания.Активно);
АктивныеЗадания = ФоновыеЗадания.ПолучитьФоновыеЗадания(ПараметрыОтбора);
Если АктивныеЗадания.Количество() > 1 Тогда
// Одно задание уже работает (само текущее задание тоже входит в список)
// Поэтому если их больше 1, значит есть "двойник"
Возврат;
КонецЕсли;
Подводя итог, можно выделить следующие правила выбора решения:
УстановитьВремяОжидания(0) у объекта БлокировкаДанных.Processing включает в себя ID сеанса или уникальный ключ задания.Помните, что любая работа с внешними веб-сервисами должна быть вынесена за пределы транзакций записи в базу данных, чтобы не вызывать длительных блокировок всей таблицы регистра.