https://pastein.ru/t/L2T
Загрузка данных
//++Варнин А.А. Задача №72300 - переработано 27.07.26
//
// Что изменено относительно исходной версии:
// 1. Доп. реквизиты (OEMНомер, АртикулМаркетплейса) забираются одним запросом в пакете,
// а не циклом по СтрЦена.Номенклатура.ДополнительныеРеквизиты (был N+1 к БД).
// 2. Наименование и Марка приходят из запроса через ПРЕДСТАВЛЕНИЕ(),
// а не через Строка(Ссылка) в цикле отправки (тоже был N+1).
// 3. Всем временным таблицам проставлен ИНДЕКСИРОВАТЬ ПО.
// 4. Убраны обращения через точку в финальном запросе — явное соединение со справочником.
// 5. Убрано ВЫРАЗИТЬ(... КАК ЧИСЛО(15,2)) из запроса (оно усекает, а не округляет).
// Округление делается один раз в коде через Окр().
// 6. Отсекается служебная номенклатура ("не использовать") и помеченная на удаление.
// 7. Отправка в Kafka переведена на ОтправитьПакетСРезультатом (реальный статус доставки),
// с фолбэком на поштучную отправку и КОРРЕКТНОЙ проверкой кода возврата
// (0 = поставлено в очередь = успех, -1 = ошибка).
// 8. Проверяется результат ОстановитьПродюсера() — именно он подтверждает flush очереди.
// 9. Рег. задание пишет старт/финиш/итоги в журнал регистрации и падает при неуспехе.
// 10. ВыгрузитьВExcel вынесена под флаг константы, путь берётся из константы.
#Область РегламентноеЗадание_кор_ВыгрузкаЗакупочныхЦен
Процедура ВыполнитьВыгрузкуЗакупочныхЦен() Экспорт
ИмяСобытия = "Выгрузка закупочных цен";
НачалоРаботы = ТекущаяУниверсальнаяДатаВМиллисекундах();
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Информация, , , "Старт выгрузки.");
ЦеныТЗ = СформироватьТаблицуЦен();
Если ЦеныТЗ.Количество() = 0 Тогда
ТекстОшибки = "Запрос не вернул ни одной строки. Выгрузка не выполнена.";
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , , ТекстОшибки);
ВызватьИсключение ТекстОшибки;
КонецЕсли;
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Информация, , ,
СтрШаблон("Сформировано строк: %1. Время запроса: %2 мс.",
ЦеныТЗ.Количество(), ТекущаяУниверсальнаяДатаВМиллисекундах() - НачалоРаботы));
РезультатОтправки = ОтправитьВKafka(ЦеныТЗ);
ВыгрузитьВExcel(ЦеныТЗ);
ЗаписьЖурналаРегистрации(ИмяСобытия,
?(РезультатОтправки.Успех, УровеньЖурналаРегистрации.Информация, УровеньЖурналаРегистрации.Ошибка),
, , СтрШаблон("Финиш. %1 Общее время: %2 мс.",
РезультатОтправки.Текст, ТекущаяУниверсальнаяДатаВМиллисекундах() - НачалоРаботы));
// Регламентное задание должно завершиться с ошибкой, иначе о проблеме никто не узнает.
Если НЕ РезультатОтправки.Успех Тогда
ВызватьИсключение РезультатОтправки.Текст;
КонецЕсли;
КонецПроцедуры
Функция СформироватьТаблицуЦен()
ВидЦены = Справочники.ВидыЦен.НайтиПоРеквизиту("Идентификатор", "ЗакупочнаяЦена");
Если НЕ ЗначениеЗаполнено(ВидЦены) Тогда
ВызватьИсключение "Не найден вид цены с идентификатором ""ЗакупочнаяЦена"".";
КонецЕсли;
СвойствоOEM = НайтиДополнительныйРеквизит("OEMНомер");
СвойствоАртикулМП = НайтиДополнительныйРеквизит("АртикулМаркетплейсов");
Запрос = Новый Запрос(ТекстЗапросаБезНДС());
Запрос.УстановитьПараметр("ВидЦены", ВидЦены);
Запрос.УстановитьПараметр("КонецПериода", ТекущаяДатаСеанса());
Запрос.УстановитьПараметр("СвойствоOEM", СвойствоOEM);
Запрос.УстановитьПараметр("СвойствоАртикулМП", СвойствоАртикулМП);
Запрос.УстановитьПараметр("МаскаСлужебных", "%не использовать%");
ЦеныТЗ = Запрос.Выполнить().Выгрузить();
// Единственное округление за весь цикл — здесь.
Для Каждого СтрЦена Из ЦеныТЗ Цикл
СтрЦена.СредняяЦена = Окр(СтрЦена.СредняяЦена, 2, РежимОкругления.Окр15как20);
СтрЦена.Цена = Окр(СтрЦена.Цена, 2, РежимОкругления.Окр15как20);
КонецЦикла;
Возврат ЦеныТЗ;
КонецФункции
// Ищет вид дополнительного реквизита по идентификатору для формул.
// Возвращает пустую ссылку, если не найден (тогда колонка просто останется пустой).
//
Функция НайтиДополнительныйРеквизит(ИдентификаторДляФормул)
Свойство = ПланыВидовХарактеристик.ДополнительныеРеквизитыИСведения.НайтиПоРеквизиту(
"ИдентификаторДляФормул", ИдентификаторДляФормул);
Если НЕ ЗначениеЗаполнено(Свойство) Тогда
ЗаписьЖурналаРегистрации("Выгрузка закупочных цен",
УровеньЖурналаРегистрации.Предупреждение, , ,
"Не найден дополнительный реквизит с идентификатором """ + ИдентификаторДляФормул + """.");
КонецЕсли;
Возврат Свойство;
КонецФункции
Функция ТекстЗапросаБезНДС()
Возврат
"ВЫБРАТЬ
| ОстаткиТоваров.Номенклатура КАК Номенклатура
|ПОМЕСТИТЬ ВТ_Остатки
|ИЗ
| РегистрНакопления.ТоварыНаСкладах.Остатки(&КонецПериода, ) КАК ОстаткиТоваров
|ГДЕ
| ОстаткиТоваров.ВНаличииОстаток > 0
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура
|;
|
|////////////////////////////////////////////////////////////////////////////////
|// Пары ""документ-основание + номенклатура"", по которым цена реально установилась.
|ВЫБРАТЬ
| УстановкаЦен.ДокументОснование КАК Основание,
| ЦеныНоменклатуры25.Номенклатура КАК Номенклатура,
| МАКСИМУМ(ЦеныНоменклатуры25.Период) КАК ПериодУстановки
|ПОМЕСТИТЬ ВТ_ДокУстановкаЦен
|ИЗ
| РегистрСведений.ЦеныНоменклатуры25 КАК ЦеныНоменклатуры25
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ Документ.УстановкаЦенНоменклатуры КАК УстановкаЦен
| ПО ЦеныНоменклатуры25.Регистратор = УстановкаЦен.Ссылка
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ ВТ_Остатки КАК ВТ_Остатки
| ПО ЦеныНоменклатуры25.Номенклатура = ВТ_Остатки.Номенклатура
|ГДЕ
| УстановкаЦен.Проведен
| И УстановкаЦен.ДокументОснование ССЫЛКА Документ.ПриобретениеТоваровУслуг
| И ЦеныНоменклатуры25.ВидЦены = &ВидЦены
| И ЦеныНоменклатуры25.Цена > 0
|
|СГРУППИРОВАТЬ ПО
| УстановкаЦен.ДокументОснование,
| ЦеныНоменклатуры25.Номенклатура
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура,
| Основание
|;
|
|////////////////////////////////////////////////////////////////////////////////
|// Суммы и количества по документу поступления в разрезе номенклатуры.
|ВЫБРАТЬ
| ДокТовары.Номенклатура КАК Номенклатура,
| Док.Ссылка КАК Ссылка,
| ВТ_ДокУстановкаЦен.ПериодУстановки КАК ПериодУстановки,
| СУММА(ВЫБОР
| КОГДА Док.ЦенаВключаетНДС
| ТОГДА ДокТовары.Сумма - ДокТовары.СуммаНДС
| ИНАЧЕ ДокТовары.Сумма
| КОНЕЦ) КАК СуммаБезНДС,
| СУММА(ДокТовары.Количество) КАК Количество
|ПОМЕСТИТЬ ВТ_ПоДокументам
|ИЗ
| Документ.ПриобретениеТоваровУслуг.Товары КАК ДокТовары
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ Документ.ПриобретениеТоваровУслуг КАК Док
| ПО ДокТовары.Ссылка = Док.Ссылка
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ ВТ_ДокУстановкаЦен КАК ВТ_ДокУстановкаЦен
| ПО Док.Ссылка = ВТ_ДокУстановкаЦен.Основание
| И ДокТовары.Номенклатура = ВТ_ДокУстановкаЦен.Номенклатура
|ГДЕ
| Док.Проведен
| И ДокТовары.Количество > 0
| И ДокТовары.Сумма > 0
|
|СГРУППИРОВАТЬ ПО
| ДокТовары.Номенклатура,
| Док.Ссылка,
| ВТ_ДокУстановкаЦен.ПериодУстановки
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура
|;
|
|////////////////////////////////////////////////////////////////////////////////
|// Цена за единицу по каждому документу. Округление НЕ делаем — оно в коде.
|ВЫБРАТЬ
| ВТ_ПоДокументам.Номенклатура КАК Номенклатура,
| ВТ_ПоДокументам.Ссылка КАК Ссылка,
| ВТ_ПоДокументам.ПериодУстановки КАК ПериодУстановки,
| ВТ_ПоДокументам.СуммаБезНДС / ВТ_ПоДокументам.Количество КАК ЦенаДокумента
|ПОМЕСТИТЬ ВТ_ЦеныДокументов
|ИЗ
| ВТ_ПоДокументам КАК ВТ_ПоДокументам
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура,
| ПериодУстановки
|;
|
|////////////////////////////////////////////////////////////////////////////////
|// Средняя закупочная: простое среднее цен документов (по постановке Шарова).
|ВЫБРАТЬ
| ВТ_ЦеныДокументов.Номенклатура КАК Номенклатура,
| СРЕДНЕЕ(ВТ_ЦеныДокументов.ЦенаДокумента) КАК СредняяЦена,
| КОЛИЧЕСТВО(ВТ_ЦеныДокументов.Ссылка) КАК КоличествоДокументов,
| МАКСИМУМ(ВТ_ЦеныДокументов.ПериодУстановки) КАК ПериодУстановки
|ПОМЕСТИТЬ ВТ_СредняяЦена
|ИЗ
| ВТ_ЦеныДокументов КАК ВТ_ЦеныДокументов
|
|СГРУППИРОВАТЬ ПО
| ВТ_ЦеныДокументов.Номенклатура
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура,
| ПериодУстановки
|;
|
|////////////////////////////////////////////////////////////////////////////////
|// Последняя цена — среднее по документам с максимальным периодом установки.
|ВЫБРАТЬ
| ВТ_ЦеныДокументов.Номенклатура КАК Номенклатура,
| СРЕДНЕЕ(ВТ_ЦеныДокументов.ЦенаДокумента) КАК ТекущаяЦенаБезНДС
|ПОМЕСТИТЬ ВТ_ПоследняяЦена
|ИЗ
| ВТ_ЦеныДокументов КАК ВТ_ЦеныДокументов
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ ВТ_СредняяЦена КАК ВТ_СредняяЦена
| ПО ВТ_ЦеныДокументов.Номенклатура = ВТ_СредняяЦена.Номенклатура
| И ВТ_ЦеныДокументов.ПериодУстановки = ВТ_СредняяЦена.ПериодУстановки
|
|СГРУППИРОВАТЬ ПО
| ВТ_ЦеныДокументов.Номенклатура
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура
|;
|
|////////////////////////////////////////////////////////////////////////////////
|// Доп. реквизиты одним запросом вместо цикла по объектам.
|// ВЫРАЗИТЬ КАК СТРОКА(150) нужен, чтобы МАКСИМУМ() отработал по составному типу
|// поля Значение. Если реальная длина реквизита больше — увеличить число.
|ВЫБРАТЬ
| ДопРеквизиты.Ссылка КАК Номенклатура,
| МАКСИМУМ(ВЫБОР
| КОГДА ДопРеквизиты.Свойство = &СвойствоOEM
| ТОГДА ВЫРАЗИТЬ(ДопРеквизиты.Значение КАК СТРОКА(150))
| ИНАЧЕ """"
| КОНЕЦ) КАК OEMНомер,
| МАКСИМУМ(ВЫБОР
| КОГДА ДопРеквизиты.Свойство = &СвойствоАртикулМП
| ТОГДА ВЫРАЗИТЬ(ДопРеквизиты.Значение КАК СТРОКА(150))
| ИНАЧЕ """"
| КОНЕЦ) КАК АртикулМаркетплейса
|ПОМЕСТИТЬ ВТ_ДопРеквизиты
|ИЗ
| Справочник.Номенклатура.ДополнительныеРеквизиты КАК ДопРеквизиты
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ ВТ_Остатки КАК ВТ_Остатки
| ПО ДопРеквизиты.Ссылка = ВТ_Остатки.Номенклатура
|ГДЕ
| ДопРеквизиты.Свойство В (&СвойствоOEM, &СвойствоАртикулМП)
|
|СГРУППИРОВАТЬ ПО
| ДопРеквизиты.Ссылка
|
|ИНДЕКСИРОВАТЬ ПО
| Номенклатура
|;
|
|////////////////////////////////////////////////////////////////////////////////
|ВЫБРАТЬ
| Ном.Ссылка КАК Номенклатура,
| ПРЕДСТАВЛЕНИЕ(Ном.Ссылка) КАК Наименование,
| Ном.Код КАК Код,
| Ном.Артикул КАК Артикул,
| ПРЕДСТАВЛЕНИЕ(Ном.Марка) КАК Марка,
| ВТ_СредняяЦена.СредняяЦена КАК СредняяЦена,
| ВТ_СредняяЦена.КоличествоДокументов КАК КоличествоДокументов,
| ВТ_ПоследняяЦена.ТекущаяЦенаБезНДС КАК Цена,
| ЕСТЬNULL(ВТ_ДопРеквизиты.OEMНомер, """") КАК OEMНомер,
| ЕСТЬNULL(ВТ_ДопРеквизиты.АртикулМаркетплейса, """") КАК АртикулМаркетплейса
|ИЗ
| ВТ_СредняяЦена КАК ВТ_СредняяЦена
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ ВТ_ПоследняяЦена КАК ВТ_ПоследняяЦена
| ПО ВТ_СредняяЦена.Номенклатура = ВТ_ПоследняяЦена.Номенклатура
| ВНУТРЕННЕЕ СОЕДИНЕНИЕ Справочник.Номенклатура КАК Ном
| ПО ВТ_СредняяЦена.Номенклатура = Ном.Ссылка
| ЛЕВОЕ СОЕДИНЕНИЕ ВТ_ДопРеквизиты КАК ВТ_ДопРеквизиты
| ПО ВТ_СредняяЦена.Номенклатура = ВТ_ДопРеквизиты.Номенклатура
|ГДЕ
| НЕ Ном.ПометкаУдаления
| И ВТ_СредняяЦена.СредняяЦена > 0
| И НЕ Ном.Наименование ПОДОБНО &МаскаСлужебных
| // Если решите отсекать неторговые позиции (сканеры, ТСД, светильники) —
| // раскомментировать и передать параметр:
| // И Ном.ВидНоменклатуры В (&ВидыНоменклатурыКВыгрузке)
|
|УПОРЯДОЧИТЬ ПО
| Код";
КонецФункции
// Формирует и отправляет пакет сообщений в Kafka.
//
// Возвращаемое значение:
// Структура - Отправлено, Ошибок, Успех, Текст
//
Функция ОтправитьВKafka(ЦеныТЗ)
Сообщения = Новый Массив;
Для Каждого СтрЦена Из ЦеныТЗ Цикл
Элемент = Новый Структура;
Элемент.Вставить("name", СокрЛП(СтрЦена.Наименование));
Элемент.Вставить("code", СокрЛП(СтрЦена.Код));
Элемент.Вставить("brand", СокрЛП(СтрЦена.Марка));
Элемент.Вставить("article", СокрЛП(СтрЦена.Артикул));
Элемент.Вставить("article_mp", СокрЛП(СтрЦена.АртикулМаркетплейса));
Элемент.Вставить("oem", СокрЛП(СтрЦена.OEMНомер));
Элемент.Вставить("last_price", СтрЦена.Цена);
Элемент.Вставить("avg_price", СтрЦена.СредняяЦена);
// Раскомментировать, если сторона калькулятора готова принимать доп. поля:
// Элемент.Вставить("docs_count", СтрЦена.КоличествоДокументов);
// Элемент.Вставить("calculated_at", Формат(ТекущаяДатаСеанса(), "ДФ=yyyy-MM-ddTHH:mm:ss"));
Сообщение = Новый Структура("Ключ, Тело",
Строка(СтрЦена.Номенклатура.УникальныйИдентификатор()),
ОбщегоНазначения.ЗначениеВJSON(Элемент));
Сообщения.Добавить(Сообщение);
КонецЦикла;
Возврат кор_HTTPКоннектор.ОтправитьВКафку(Сообщения);
КонецФункции
Процедура ВыгрузитьВExcel(ЦеныТЗ)
// Отладочная выгрузка. Включается константой, путь тоже из константы,
// чтобы не хардкодить каталог конкретного сервера.
Если НЕ Константы.кор_ВыгружатьЦеныВФайл.Получить() Тогда
Возврат;
КонецЕсли;
Каталог = СокрЛП(Константы.кор_КаталогВыгрузкиЦен.Получить());
Если НЕ ЗначениеЗаполнено(Каталог) Тогда
ЗаписьЖурналаРегистрации("Выгрузка закупочных цен",
УровеньЖурналаРегистрации.Предупреждение, , ,
"Не заполнена константа кор_КаталогВыгрузкиЦен, файл не сохранён.");
Возврат;
КонецЕсли;
Попытка
ТабДок = Новый ТабличныйДокумент;
Построитель = Новый ПостроительОтчета;
Построитель.ИсточникДанных = Новый ОписаниеИсточникаДанных(ЦеныТЗ);
Построитель.Выполнить();
Построитель.Вывести(ТабДок);
ИмяФайла = Каталог + ?(Прав(Каталог, 1) = ПолучитьРазделительПути(), "", ПолучитьРазделительПути())
+ "цены_" + Формат(ТекущаяДатаСеанса(), "ДФ=yyyyMMdd_HHmmss") + ".xlsx";
ТабДок.Записать(ИмяФайла, ТипФайлаТабличногоДокумента.XLSX);
Исключение
// Отладочная выгрузка не должна ронять регламент, но и молчать не должна.
ЗаписьЖурналаРегистрации("Выгрузка закупочных цен",
УровеньЖурналаРегистрации.Предупреждение, , ,
"Не удалось сохранить файл: " + ПодробноеПредставлениеОшибки(ИнформацияОбОшибке()));
КонецПопытки;
КонецПроцедуры
#КонецОбласти
// https://github.com/NuclearAPK/Simple-Kafka_Adapter
// Общий модуль: кор_HTTPКоннектор
//
// Настройки хранятся в константах:
// кор_KafkaБрокеры (Строка) - "host:port,host:port,..."
// кор_KafkaТопик (Строка) - имя топика
// кор_KafkaКаталогЛогов (Строка) - каталог логов компоненты
//
#Область Кафка
#Область ПрограммныйИнтерфейс
Функция ПолучитьСтруктуруПодключения() Экспорт
СтруктураПодключения = Новый Структура;
СтруктураПодключения.Вставить("Брокеры", СокрЛП(Константы.кор_KafkaБрокеры.Получить()));
СтруктураПодключения.Вставить("ИмяТопика", СокрЛП(Константы.кор_KafkaТопик.Получить()));
СтруктураПодключения.Вставить("КаталогЛогов", СокрЛП(Константы.кор_KafkaКаталогЛогов.Получить()));
Возврат СтруктураПодключения;
КонецФункции
// Отправляет поток сообщений в Kafka.
//
// Параметры:
// Сообщения - Массив структур - каждый элемент: Ключ (Строка), Тело (Строка JSON)
//
// Возвращаемое значение:
// Структура - Отправлено (Число), Ошибок (Число), Успех (Булево), Текст (Строка)
//
Функция ОтправитьВКафку(Сообщения) Экспорт
ИмяСобытия = "Kafka. Отправка цен";
Результат = Новый Структура("Отправлено, Ошибок, Успех, Текст", 0, 0, Ложь, "");
Если ТипЗнч(Сообщения) <> Тип("Массив") ИЛИ Сообщения.Количество() = 0 Тогда
Результат.Текст = "Нет данных для отправки.";
Возврат Результат;
КонецЕсли;
Подключение = ПолучитьСтруктуруПодключения();
Если НЕ ЗначениеЗаполнено(Подключение.Брокеры) Тогда
Результат.Текст = "Не заполнена константа кор_KafkaБрокеры.";
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , , Результат.Текст);
Возврат Результат;
КонецЕсли;
Если НЕ ЗначениеЗаполнено(Подключение.ИмяТопика) Тогда
Результат.Текст = "Не заполнена константа кор_KafkaТопик.";
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , , Результат.Текст);
Возврат Результат;
КонецЕсли;
Компонента = СоздатьКомпонентуКафкаКлиент();
Если Компонента = Неопределено Тогда
Результат.Текст = "Не удалось создать компоненту Kafka.";
Возврат Результат;
КонецЕсли;
Если ЗначениеЗаполнено(Подключение.КаталогЛогов) Тогда
Попытка
Компонента.КаталогЛогов = Подключение.КаталогЛогов;
Исключение
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Предупреждение, , ,
"Не удалось установить КаталогЛогов: " + ПодробноеПредставлениеОшибки(ИнформацияОбОшибке()));
КонецПопытки;
КонецЕсли;
// Продюсер при остановке ждёт доставки всех сообщений из очереди.
// Дефолт 20 сек — на нашем объёме хватает, но запас лишним не будет.
Попытка
Компонента.УстановитьТаймаутОчисткиПродюсера(60000);
Исключение
// Метода нет в старых сборках — не критично.
КонецПопытки;
Если НЕ Компонента.ИнициализироватьПродюсера(Подключение.Брокеры) Тогда
Результат.Текст = "Не удалось инициализировать продюсера. Брокеры: " + Подключение.Брокеры
+ ". " + Компонента.ПолучитьСообщениеОбОшибке();
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , , Результат.Текст);
Компонента = Неопределено;
Возврат Результат;
КонецЕсли;
Всего = Сообщения.Количество();
Попытка
ОтправитьПакетами(Компонента, Сообщения, Подключение.ИмяТопика, Результат, ИмяСобытия);
Исключение
Результат.Текст = "Прервано на " + (Результат.Отправлено + Результат.Ошибок) + " из " + Всего
+ ". " + ПодробноеПредставлениеОшибки(ИнформацияОбОшибке());
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , , Результат.Текст);
Попытка Компонента.ОстановитьПродюсера(); Исключение КонецПопытки;
Компонента = Неопределено;
Возврат Результат;
КонецПопытки;
// Именно ОстановитьПродюсера подтверждает, что очередь реально ушла брокеру.
ОчередьОчищена = Компонента.ОстановитьПродюсера();
Если НЕ ОчередьОчищена Тогда
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
"Продюсер остановлен с ошибкой (часть сообщений могла не дойти): "
+ Компонента.ПолучитьСообщениеОбОшибке());
КонецЕсли;
Компонента = Неопределено;
Результат.Успех = (Результат.Ошибок = 0) И ОчередьОчищена;
Результат.Текст = СтрШаблон("Отправлено: %1, ошибок: %2, всего: %3.",
Результат.Отправлено, Результат.Ошибок, Всего);
ЗаписьЖурналаРегистрации(ИмяСобытия,
?(Результат.Успех, УровеньЖурналаРегистрации.Информация, УровеньЖурналаРегистрации.Предупреждение),
, , Результат.Текст);
Возврат Результат;
КонецФункции
#КонецОбласти
#Область СлужебныеПроцедурыИФункции
// Отправляет сообщения пачками через ОтправитьПакетСРезультатом.
// Если метода нет (старая сборка компоненты) — уходит на поштучную отправку.
//
Процедура ОтправитьПакетами(Компонента, Сообщения, Топик, Результат, ИмяСобытия)
РазмерПачки = 1000;
Пачка = Новый Массив;
Для Каждого ЭлементСообщения Из Сообщения Цикл
Ключ = ?(ЭлементСообщения.Свойство("Ключ"), Строка(ЭлементСообщения.Ключ), "");
Тело = ?(ЭлементСообщения.Свойство("Тело"), Строка(ЭлементСообщения.Тело), "");
Если НЕ ЗначениеЗаполнено(Тело) Тогда
Результат.Ошибок = Результат.Ошибок + 1;
Продолжить;
КонецЕсли;
СообщениеКомпоненты = Новый Структура;
СообщениеКомпоненты.Вставить("message", Тело);
Если ЗначениеЗаполнено(Ключ) Тогда
СообщениеКомпоненты.Вставить("key", Ключ);
КонецЕсли;
Пачка.Добавить(СообщениеКомпоненты);
Если Пачка.Количество() >= РазмерПачки Тогда
ОтправитьОднуПачку(Компонента, Пачка, Топик, Результат, ИмяСобытия);
Пачка = Новый Массив;
КонецЕсли;
КонецЦикла;
Если Пачка.Количество() > 0 Тогда
ОтправитьОднуПачку(Компонента, Пачка, Топик, Результат, ИмяСобытия);
КонецЕсли;
КонецПроцедуры
Процедура ОтправитьОднуПачку(Компонента, Пачка, Топик, Результат, ИмяСобытия)
ЗаписьJSON = Новый ЗаписьJSON;
ЗаписьJSON.УстановитьСтроку();
ЗаписатьJSON(ЗаписьJSON, Пачка);
ПачкаJSON = ЗаписьJSON.Закрыть();
ОтветJSON = "";
ЕстьПакетныйМетод = Истина;
Попытка
ОтветJSON = Компонента.ОтправитьПакетСРезультатом(ПачкаJSON, Топик, 60000);
Исключение
ЕстьПакетныйМетод = Ложь;
КонецПопытки;
Если НЕ ЕстьПакетныйМетод Тогда
ОтправитьПоштучно(Компонента, Пачка, Топик, Результат, ИмяСобытия);
Возврат;
КонецЕсли;
Если ПустаяСтрока(ОтветJSON) Тогда
Результат.Ошибок = Результат.Ошибок + Пачка.Количество();
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
"Пакет не отправлен: " + Компонента.ПолучитьСообщениеОбОшибке());
Возврат;
КонецЕсли;
ЧтениеJSON = Новый ЧтениеJSON;
ЧтениеJSON.УстановитьСтроку(ОтветJSON);
ОтветПакета = ПрочитатьJSON(ЧтениеJSON);
ЧтениеJSON.Закрыть();
// Компонента возвращает счётчики строками, поэтому явное приведение.
Доставлено = Число(ОтветПакета["success_count"]);
НеДоставлено = Число(ОтветПакета["failed_count"]);
Результат.Отправлено = Результат.Отправлено + Доставлено;
Результат.Ошибок = Результат.Ошибок + НеДоставлено;
Если НеДоставлено > 0 Тогда
ЗаписатьОшибкиПакета(ОтветПакета, ИмяСобытия);
КонецЕсли;
КонецПроцедуры
Процедура ЗаписатьОшибкиПакета(ОтветПакета, ИмяСобытия)
СтрокиОшибок = Новый Массив;
Для Каждого СтрокаРезультата Из ОтветПакета["results"] Цикл
Если СтрокаРезультата["delivered"] = "true" Тогда
Продолжить;
КонецЕсли;
СтрокиОшибок.Добавить(СтрШаблон("Ключ: %1. %2",
СтрокаРезультата["key"], СтрокаРезультата["error"]));
// Журнал не резиновый — 50 примеров достаточно, остальное в логе компоненты.
Если СтрокиОшибок.Количество() >= 50 Тогда
Прервать;
КонецЕсли;
КонецЦикла;
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
"Недоставленные сообщения:" + Символы.ПС + СтрСоединить(СтрокиОшибок, Символы.ПС));
КонецПроцедуры
// Фолбэк на поштучную асинхронную отправку.
// ВАЖНО: у свежих версий компоненты ОтправитьСообщение возвращает ЧИСЛО:
// 0 - сообщение помещено в очередь (это УСПЕХ),
// -1 - ошибка помещения в очередь.
// Старая проверка "Если РезультатОтправки Тогда" считала все нули ошибками.
//
Процедура ОтправитьПоштучно(Компонента, Пачка, Топик, Результат, ИмяСобытия)
Для Каждого СообщениеКомпоненты Из Пачка Цикл
Ключ = "";
СообщениеКомпоненты.Свойство("key", Ключ);
Тело = СообщениеКомпоненты["message"];
РезультатОтправки = Компонента.ОтправитьСообщение(Тело, Топик, -1, Ключ);
Если РезультатОтправки = -1 Тогда
Результат.Ошибок = Результат.Ошибок + 1;
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
"Не поставлено в очередь. Ключ: " + Ключ + ". " + Компонента.ПолучитьСообщениеОбОшибке());
Иначе
Результат.Отправлено = Результат.Отправлено + 1;
КонецЕсли;
КонецЦикла;
КонецПроцедуры
// Создаёт объект внешней компоненты Kafka из общего макета.
//
Функция СоздатьКомпонентуКафкаКлиент()
ИмяСобытия = "Kafka. Подключение компоненты";
Попытка
Подключено = ПодключитьВнешнююКомпоненту(
"ОбщийМакет.SimpleKafkaAdapter64",
"KafkaClient",
ТипВнешнейКомпоненты.Native,
ТипПодключенияВнешнейКомпоненты.Изолированно);
Исключение
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
ПодробноеПредставлениеОшибки(ИнформацияОбОшибке()));
Возврат Неопределено;
КонецПопытки;
Если НЕ Подключено Тогда
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
"Не удалось подключить компоненту из общего макета SimpleKafkaAdapter64.");
Возврат Неопределено;
КонецЕсли;
Попытка
Компонента = Новый("AddIn.KafkaClient.simpleKafka1C");
Исключение
ЗаписьЖурналаРегистрации(ИмяСобытия, УровеньЖурналаРегистрации.Ошибка, , ,
"Компонента подключилась, но объект не создался: "
+ ПодробноеПредставлениеОшибки(ИнформацияОбОшибке()));
Возврат Неопределено;
КонецПопытки;
Возврат Компонента;
КонецФункции
#КонецОбласти
#КонецОбласти
