Вступление: когда JSON перестал справляться
Представьте, что вы отправляете посылку курьерской службой. JSON — это как если бы вы упаковали каждый предмет в отдельную коробку с подробным описанием на каждой стороне: "Это кружка. Цвет: белый. Материал: керамика. Объем: 300мл." Удобно для понимания, но занимает много места.
AVRO и Protobuf — это как компактный штрих-код на минималистичной упаковке. Вся информация там есть, но места занимает в разы меньше, а читается мгновенно специальным сканером.
В мире 1С мы долгое время жили в парадигме "текст решает всё": XML для обмена, JSON для API, CSV для выгрузок. Это работало, пока объемы данных были скромными, а интеграции — не слишком частыми. Но современная архитектура требует другого подхода.
Simple Kafka Connector — это не просто конвертер форматов. Это полноценный клиент Apache Kafka, который включает в себя всю функциональность профессиональной работы с распределённой шиной сообщений: транзакции с гарантией Exactly-Once, управление consumer groups, пакетную отправку, Admin API для управления топиками и кластером. Поддержка AVRO и Protobuf — это мощная надстройка над уже зрелым и проверенным в production решением, которое успешно используется в высоконагруженных системах.
Актуальность в контексте микросервисной архитектуры
Проблемы текстовых форматов в enterprise-окружении
В классической архитектуре "1С + несколько внешних систем" JSON справлялся отлично. Но когда ваша инфраструктура вырастает до десятков микросервисов, обменивающихся миллионами сообщений в день через Apache Kafka, начинаются проблемы:
1. Размер имеет значение
Типичное JSON-сообщение с информацией о заказе:
{
"order_id": "ORD-2026-00001234",
"customer_id": "CUST-567890",
"amount": 15420.50,
"currency": "RUB",
"status": "pending",
"created_at": "2026-01-11T10:30:00Z"
}
Размер: ~180 байт
То же сообщение в Protobuf: ~45 байт (экономия 75%)
Когда через Kafka проходит 10 миллионов таких сообщений в день, разница между 1.8 ГБ и 450 МБ трафика становится критичной для стоимости инфраструктуры и скорости обработки.
2. Скорость сериализации
JSON требует парсинга текста, проверки синтаксиса, преобразования типов. Бинарные форматы читаются напрямую в память с минимальными преобразованиями. На практике это даёт разницу в 5-10 раз в скорости обработки.
3. Строгая типизация и эволюция схем
В JSON-мире мы часто сталкиваемся с проблемами:
- "Почему в одном сообщении amount — строка, а в другом — число?"
- "Мы добавили новое поле, и старая версия сервиса упала!"
- "Как понять, какая структура данных ожидается на входе?"
AVRO и Protobuf решают это через схемы — формальные описания структуры данных, которые:
- Гарантируют типобезопасность
- Обеспечивают обратную и прямую совместимость при изменениях
- Служат документацией для разработчиков
Место 1С в микросервисной экосистеме
Современная enterprise-система выглядит так:
- 1С — система учета и управления (ERP-ядро)
- Apache Kafka — шина обмена сообщениями
- Микросервисы — специализированные сервисы (аналитика, ML, интеграции, нотификации)
До появления поддержки AVRO и Protobuf в Simple Kafka Connector разработчики сталкивались с дилеммой:
- Либо использовать JSON и терять в производительности
- Либо создавать промежуточные сервисы-конвертеры (что увеличивает задержки и усложняет архитектуру)
Теперь 1С может напрямую "говорить" на языке современных микросервисов, становясь полноправным участником событийно-ориентированной архитектуры.
AVRO vs Protobuf: выбираем правильный инструмент
Оба формата решают схожие задачи, но имеют разную философию. Выбор между ними — это не вопрос "что лучше", а вопрос "что подходит для вашего случая".
Apache AVRO: самодокументируемые данные
Философия: Схема всегда с данными
AVRO хранит схему прямо в файле/сообщении. Это как если бы к каждой посылке прилагалась инструкция по распаковке.
Когда использовать AVRO:
Частые изменения схем Когда ваша модель данных активно развивается, и нужна максимальная гибкость. AVRO поддерживает прямую и обратную совместимость "из коробки".
Аналитические системы и Data Lake Когда данные сохраняются надолго, и через год нужно будет прочитать, что именно было записано. Схема в файле — это самодокументирование.
Динамические данные Когда структура данных может варьироваться, и нужна возможность работы со схемой в runtime.
Интеграция с экосистемой Hadoop AVRO — стандарт де-факто для Hadoop, Spark, Hive.
Пример сценария:
Задача: Выгрузка документов из 1С в аналитическое хранилище
Частота: 1 раз в сутки
Объем: 1 млн документов
Особенность: Модель данных меняется раз в квартал (новые реквизиты)
Решение: AVRO
Причина: Схема сохраняется вместе с данными, старые выгрузки
остаются читаемыми, новые поля добавляются без проблем.
Protocol Buffers: максимальная производительность
Философия: Схема заранее известна всем участникам
Protobuf требует, чтобы все системы заранее знали схему данных. Это как согласованный протокол связи между космическими кораблями — быстро, но требует предварительной договоренности.
Когда использовать Protobuf:
Высоконагруженные системы Когда счет идёт на миллионы сообщений в минуту, и каждая миллисекунда важна.
Устоявшиеся схемы Когда структура данных стабильна и меняется редко.
Минимальный размер сообщений Protobuf на 20-30% компактнее AVRO. Критично для IoT, мобильных приложений, оплаты трафика.
Интеграция с современными микросервисами Protobuf — стандарт для gRPC, используется в Google, Netflix, Square.
Пример сценария:
Задача: Синхронизация остатков в реальном времени
Частота: 1000 сообщений в секунду
Объем: Небольшие структуры (артикул, склад, остаток)
Особенность: Схема стабильна годами
Решение: Protobuf
Причина: Максимальная скорость обработки, минимальный размер,
схема меняется редко.
Сравнительная таблица
| Критерий |
AVRO |
Protobuf |
JSON (для сравнения) |
| Размер сообщения |
Компактный (~50% от JSON) |
Очень компактный (~30% от JSON) |
Базовый (100%) |
| Скорость обработки |
Быстрая (в 5-7 раз быстрее JSON) |
Очень быстрая (в 8-10 раз быстрее JSON) |
Базовая |
| Схема в данных |
Да (самодокументирование) |
 Нет (нужно знать заранее) |
Нет |
| Эволюция схемы |
 Прямая и обратная совместимость |
Обратная совместимость |
Зависит от реализации |
| Читаемость без инструментов |
 Бинарный формат |
Бинарный формат |
Текстовый формат |
| Экосистема |
Hadoop, Spark, Kafka, Starrocks |
gRPC, Microservices APIs, Cloud Native |
Универсальная |
| Сложность внедрения |
Средняя |
Средняя |
Низкая |
| Использование в 1С |
 Поддерживается |
Поддерживается |
Встроенная поддержка |
Золотое правило выбора
Используйте AVRO, если:
- Данные хранятся долго (архивы, аналитические хранилища)
- Схема часто меняется
- Важна самодокументируемость
Используйте Protobuf, если:
- Нужна максимальная производительность
- Схема стабильна
- Интегрируетесь с gRPC-сервисами
Оставайтесь на JSON, если:
- Объемы данных небольшие (< 100k сообщений/день)
- Важна читаемость и отладка
- Интегрируетесь с legacy-системами
Практика: примеры использования в 1С
Пример 1: Отправка данных в формате AVRO
Представим задачу: отправка информации о проведенных документах "Реализация товаров и услуг" в аналитическое хранилище через Kafka.
Шаг 1: Определяем схему AVRO
роцедура ИнициализацияСхемыAVRO(Компонента)
ИмяСхемыJSON = "sales_document";
// Определяем структуру данных
СхемаJSON = "{
| ""type"": ""record"",
| ""name"": ""sales_document"",
| ""fields"": [
| {
| ""name"": ""document_id"",
| ""type"": {
| ""type"": ""string"",
| ""logicalType"": ""uuid""
| }
| },
| {
| ""name"": ""document_number"",
| ""type"": ""string""
| },
| {
| ""name"": ""document_date"",
| ""type"": ""long"",
| ""logicalType"": ""timestamp-millis""
| },
| {
| ""name"": ""customer_id"",
| ""type"": [""null"", ""string""]
| },
| {
| ""name"": ""amount"",
| ""type"": ""double""
| },
| {
| ""name"": ""currency"",
| ""type"": ""string""
| },
| {
| ""name"": ""line_count"",
| ""type"": ""int""
| }
| ]
|}";
// Сохраняем схему в компоненте
Результат = Компонента.СохранитьСхемуAVRO(ИмяСхемыJSON, СхемаJSON);
Если Не Результат Тогда
ВызватьИсключение "Ошибка сохранения схемы AVRO: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
КонецПроцедуры
Шаг 2: Формируем и отправляем данные
Процедура ОтправитьДокументыВAVRO(МассивДокументов, Компонента)
// AVRO ожидает данные в формате колонок (columnar format)
// Это обеспечивает лучшее сжатие и производительность
МассивИдентификаторов = Новый Массив;
МассивНомеров = Новый Массив;
МассивДат = Новый Массив;
МассивКонтрагентов = Новый Массив;
МассивСумм = Новый Массив;
МассивВалют = Новый Массив;
МассивСтрок = Новый Массив;
Для Каждого Документ Из МассивДокументов Цикл
МассивИдентификаторов.Добавить(Строка(Документ.УникальныйИдентификатор()));
МассивНомеров.Добавить(Документ.Номер);
МассивДат.Добавить((Документ.Дата - Дата(1970, 1, 1)) * 1000); // Unix timestamp в миллисекундах
Если ЗначениеЗаполнено(Документ.Контрагент) Тогда
МассивКонтрагентов.Добавить(Строка(Документ.Контрагент.УникальныйИдентификатор()));
Иначе
МассивКонтрагентов.Добавить(Null); // AVRO поддерживает null для Union-типов
КонецЕсли;
МассивСумм.Добавить(Документ.СуммаДокумента);
МассивВалют.Добавить(Строка(Документ.ВалютаДокумента));
МассивСтрок.Добавить(Документ.Товары.Количество());
КонецЦикла;
// Формируем JSON для AVRO (колоночный формат)
ДанныеJSON = Новый Структура;
ДанныеJSON.Вставить("document_id", МассивИдентификаторов);
ДанныеJSON.Вставить("document_number", МассивНомеров);
ДанныеJSON.Вставить("document_date", МассивДат);
ДанныеJSON.Вставить("customer_id", МассивКонтрагентов);
ДанныеJSON.Вставить("amount", МассивСумм);
ДанныеJSON.Вставить("currency", МассивВалют);
ДанныеJSON.Вставить("line_count", МассивСтрок);
ЗаписьJSON = Новый ЗаписьJSON;
ЗаписьJSON.УстановитьСтроку();
ЗаписатьJSON(ЗаписьJSON, ДанныеJSON);
СтрокаJSON = ЗаписьJSON.Закрыть();
// Преобразуем в формат AVRO
Результат = Компонента.ПреобразоватьВФорматAVRO(СтрокаJSON, "sales_document");
Если Не Результат Тогда
ВызватьИсключение "Ошибка преобразования в AVRO: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
// Отправляем в Kafka
Топик = "sales.documents.v1";
Ключ = Формат(ТекущаяДата(), "ДФ=yyyyMMdd"); // Партиционирование по дате
РезультатОтправки = Компонента.ОтправитьСообщениеAVRO(Топик, , Ключ);
Если РезультатОтправки = -1 Тогда
ВызватьИсключение "Ошибка отправки AVRO: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
Сообщить("Отправлено документов в AVRO: " + МассивДокументов.Количество());
КонецПроцедуры
Результат: Вместо отправки отдельного JSON-сообщения на каждый документ (180 байт × 1000 документов = 180 КБ), мы отправляем одно AVRO-сообщение размером ~45 КБ. Экономия трафика: 75%.
Пример 2: Получение данных в формате Protobuf
Представим обратную задачу: получение обновлений остатков из внешней системы через Kafka в формате Protobuf.
Шаг 1: Определяем схему Protobuf
Процедура ИнициализацияСхемыProtobuf(Компонента)
ИмяСхемы = "StockUpdate";
// .proto схема (синтаксис Protocol Buffers)
СхемаProto = "
|syntax = ""proto3"";
|
|message StockUpdate {
| string product_id = 1;
| string warehouse_id = 2;
| double quantity = 3;
| string unit = 4;
| int64 timestamp = 5;
| string source_system = 6;
|}
|";
Результат = Компонента.СохранитьСхемуProtobuf(ИмяСхемы, СхемаProto);
Если Не Результат Тогда
ВызватьИсключение "Ошибка сохранения схемы Protobuf: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
КонецПроцедуры
Шаг 2: Подписка и обработка сообщений
Процедура ОбработкаОстатковИзKafka(Компонента)
// Инициализируем потребителя
АдресБрокера = "kafka-cluster.company.local:9092";
ГруппаПотребителей = "1c_stock_processor";
Компонента.ИнициализироватьКонсьюмера(АдресБрокера);
Компонента.УстановитьПараметр("group.id", ГруппаПотребителей);
Компонента.УстановитьПараметр("auto.offset.reset", "earliest");
// Подписываемся на топик
Топик = "warehouse.stock.updates";
Компонента.Подписаться(Топик);
Сообщить("Ожидание сообщений из топика: " + Топик);
// Цикл обработки сообщений
КоличествоОбработанных = 0;
ТаймаутМс = 5000; // 5 секунд
Пока КоличествоОбработанных < 100 Цикл // Обработаем 100 сообщений для примера
// Получаем сообщение
Если Компонента.ПрочитатьСообщение(ТаймаутМс) Тогда
// Получаем бинарные данные Protobuf
ДанныеProtobuf = Компонента.ПолучитьДанныеСообщения(Истина); // Истина = бинарные данные
// Декодируем Protobuf в JSON
ДанныеJSON = Компонента.ДекодироватьСообщениеProtobuf(
ДанныеProtobuf,
"StockUpdate",
Истина // Возвращать JSON
);
Если ПустаяСтрока(ДанныеJSON) Тогда
ВызватьИсключение "Ошибка декодирования Protobuf: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
// Разбираем JSON
ЧтениеJSON = Новый ЧтениеJSON;
ЧтениеJSON.УстановитьСтроку(ДанныеJSON);
ДанныеОстатка = ПрочитатьJSON(ЧтениеJSON);
ЧтениеJSON.Закрыть();
// Обрабатываем данные
ОбработатьОстаток(ДанныеОстатка);
КоличествоОбработанных = КоличествоОбработанных + 1;
Иначе
// Таймаут - сообщений нет
Сообщить("Сообщений больше нет, завершение обработки.");
Прервать;
КонецЕсли;
КонецЦикла;
Сообщить("Обработано сообщений: " + КоличествоОбработанных);
КонецПроцедуры
Процедура ОбработатьОстаток(Данные)
// Находим номенклатуру и склад по UUID
Номенклатура = Справочники.Номенклатура.ПолучитьСсылку(
Новый УникальныйИдентификатор(Данные["product_id"])
);
Склад = Справочники.Склады.ПолучитьСсылку(
Новый УникальныйИдентификатор(Данные["warehouse_id"])
);
Количество = Данные["quantity"];
ВременнаяМетка = Дата(1970, 1, 1) + Данные["timestamp"] / 1000;
// Здесь логика обновления остатков в 1С
// Например, создание документа корректировки или прямое изменение регистра
Сообщить(СтрШаблон(
"Остаток обновлен: %1 на складе %2 = %3 (источник: %4)",
Номенклатура,
Склад,
Количество,
Данные["source_system"]
));
КонецПроцедуры
Результат: Обработка остатков происходит в 8-10 раз быстрее, чем при использовании JSON, благодаря бинарному формату Protobuf.
Пример 3: Микросервисная интеграция — читаем AVRO, пишем Protobuf
Рассмотрим типичный сценарий микросервисной архитектуры:
- Микросервис заказов (Order Service) отправляет новые заказы в формате AVRO в топик
orders.new (AVRO обеспечивает самодокументируемость и гибкую эволюцию схемы)
- 1С ERP читает заказы, проверяет остатки, резервирует товар, проводит бизнес-логику
- 1С ERP отправляет статус обработки в формате Protobuf в топик
orders.processing.status (Protobuf обеспечивает максимальную производительность для real-time обработки)
- Микросервисы (нотификации, логистика, биллинг) читают статусы в Protobuf для дальнейшей обработки
- Аналитическая система (Kafka Streams, Apache Spark) читает данные из обоих топиков для построения отчётов и дашбордов
Это демонстрирует, как 1С может быть центральным звеном обработки в гетерогенной микросервисной среде, работая с разными форматами данных.
Архитектура решения
Шаг 1: Схемы данных
AVRO схема для входящих заказов (от Order Service)
Процедура ИнициализацияСхемыAVRO_Заказ(Компонента)
ИмяСхемы = "order_event";
СхемаJSON = "{
| ""type"": ""record"",
| ""name"": ""order_event"",
| ""namespace"": ""com.company.orders"",
| ""fields"": [
| {
| ""name"": ""order_id"",
| ""type"": {
| ""type"": ""string"",
| ""logicalType"": ""uuid""
| },
| ""doc"": ""Уникальный идентификатор заказа""
| },
| {
| ""name"": ""customer_id"",
| ""type"": ""string""
| },
| {
| ""name"": ""order_date"",
| ""type"": ""long"",
| ""logicalType"": ""timestamp-millis""
| },
| {
| ""name"": ""items"",
| ""type"": {
| ""type"": ""array"",
| ""items"": {
| ""type"": ""record"",
| ""name"": ""order_item"",
| ""fields"": [
| {
| ""name"": ""product_id"",
| ""type"": ""string""
| },
| {
| ""name"": ""quantity"",
| ""type"": ""double""
| },
| {
| ""name"": ""price"",
| ""type"": ""double""
| }
| ]
| }
| }
| },
| {
| ""name"": ""total_amount"",
| ""type"": ""double""
| },
| {
| ""name"": ""source_system"",
| ""type"": ""string"",
| ""default"": ""order-service""
| }
| ]
|}";
Результат = Компонента.СохранитьСхемуAVRO(ИмяСхемы, СхемаJSON);
Если Не Результат Тогда
ВызватьИсключение "Ошибка инициализации AVRO схемы: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
КонецПроцедуры
Protobuf схема для статуса обработки
Процедура ИнициализацияСхемыProtobuf_Статус(Компонента)
ИмяСхемы = "OrderProcessingStatus";
СхемаProto = "
|syntax = ""proto3"";
|
|message OrderProcessingStatus {
| string order_id = 1;
| string status = 2; // RESERVED, CONFIRMED, REJECTED, SHIPPED
| int64 processed_at = 3; // Unix timestamp в миллисекундах
| string processed_by = 4; // Идентификатор 1С инстанса
| repeated ReservationItem reservations = 5;
| string rejection_reason = 6; // Причина отказа (если status = REJECTED)
| double total_reserved_amount = 7;
|}
|
|message ReservationItem {
| string product_id = 1;
| string warehouse_id = 2;
| double quantity = 3;
| double reserved_amount = 4;
|}
|";
Результат = Компонента.СохранитьСхемуProtobuf(ИмяСхемы, СхемаProto);
Если Не Результат Тогда
ВызватьИсключение "Ошибка инициализации Protobuf схемы: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
КонецПроцедуры
Шаг 2: Обработка заказов с Exactly-Once гарантиями (Read-Process-Write)
Используем транзакционный подход для обеспечения Exactly-Once семантики. Это гарантирует:
- Каждый заказ обрабатывается ровно один раз (нет дубликатов, нет потерь)
- Атомарность: либо всё выполняется (чтение, обработка в 1С, запись статуса), либо ничего
- При сбоях и перезапусках обработка продолжается с правильной позиции
Обработка заказов из Kafka (читаем AVRO, пишем Protobuf)
Процедура ОбработкаЗаказовИзKafka_ExactlyOnce()
// Инициализация компоненты
Компонента = Новый("AddIn.Integration.simpleKafka1C");
// Загружаем схемы
ИнициализацияСхемыAVRO_Заказ(Компонента);
ИнициализацияСхемыProtobuf_Статус(Компонента);
// Настраиваем consumer для Exactly-Once
ГруппаПотребителей = "1c-order-processor";
Компонента.УстановитьПараметр("group.id", ГруппаПотребителей );
Компонента.УстановитьПараметр("auto.offset.reset", "earliest");
// ВАЖНО: Отключаем автокоммит offset - будем управлять вручную через транзакции
Компонента.УстановитьПараметр("enable.auto.commit", "false");
// Настройка для чтения только committed сообщений (для полной изоляции)
Компонента.УстановитьПараметр("isolation.level", "read_committed");
// Инициализируем consumer
АдресБрокера = "kafka-cluster.company.local:9092";
Компонента.ИнициализироватьКонсьюмера(АдресБрокера);
// Инициализируем ТРАНЗАКЦИОННОГО producer
// Уникальный ID для каждого экземпляра 1С (при перезапуске используем тот же ID)
ИдентификаторТранзакции = "1c-order-processor-instance-01";
РезультатИнициализации = Компонента.ИнициализироватьТранзакционногоПродюсера(
АдресБрокера,
ИдентификаторТранзакции
);
Если Не РезультатИнициализации Тогда
ВызватьИсключение "Ошибка инициализации транзакционного продюсера: "
+ Компонента.ПолучитьСообщениеОбОшибке();
КонецЕсли;
// Подписываемся на топик с новыми заказами (от Order Service в AVRO)
ТопикВход = "orders.new";
Компонента.Подписаться(ТопикВход);
Сообщить("Запущена обработка заказов из топика: " + ТопикВход);
// Топик для отправки статусов в Protobuf
ТопикВыход = "orders.processing.status";
// Счётчики для статистики
КоличествоОбработано = 0;
КоличествоПодтверждено = 0;
КоличествоОтклонено = 0;
// Основной цикл обработки
Пока Истина Цикл
Попытка
// Читаем сообщение из Kafka (таймаут 10 секунд)
Если Компонента.ПрочитатьСообщение(10000) Тогда
// *** НАЧИНАЕМ ТРАНЗАКЦИЮ KAFKA для Exactly-Once ***
Компонента.НачатьТранзакцию();
Попытка
// Получаем AVRO данные
ДанныеAVRO = Компонента.ПолучитьДанныеСообщения(Истина); // Истина = бинарные данные
ТопикСообщения = Компонента.ПолучитьТопикСообщения();
ПартицияСообщения = Компонента.ПолучитьРазделСообщения();
ОфсетСообщения = Компонента.ПолучитьСмещениеСообщения();
// Декодируем AVRO в JSON
ДанныеJSON = Компонента.ДекодироватьСообщениеAVRO(
ДанныеAVRO,
"order_event",
Истина
);
Если ПустаяСтрока(ДанныеJSON) Тогда
ВызватьИсключение "Ошибка декодирования AVRO: "
+ Компонента.ПолучитьСообщениеОбОшибке();
КонецЕсли;
// Парсим JSON
ЧтениеJSON = Новый ЧтениеJSON;
ЧтениеJSON.УстановитьСтроку(ДанныеJSON);
ДанныеЗаказа = ПрочитатьJSON(ЧтениеJSON);
ЧтениеJSON.Закрыть();
// Обрабатываем заказ в 1С
СтатусОбработки = ОбработатьЗаказВ1С(ДанныеЗаказа);
// Формируем статус для отправки в Protobuf
СтатусJSON = СформироватьСтатусJSON(СтатусОбработки);
// Преобразуем в Protobuf
Результат = Компонента.ПреобразоватьВФорматProtobuf(
СтатусJSON,
"OrderProcessingStatus"
);
Если Не Результат Тогда
ВызватьИсключение "Ошибка преобразования в Protobuf: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
// Отправляем статус в Kafka (Protobuf)
КлючСообщения = СтатусОбработки.ИдентификаторЗаказа;
РезультатОтправки = Компонента.ОтправитьСообщениеProtobuf(
ТопикВыход,
-1, // Автоматический выбор партиции
КлючСообщения
);
Если РезультатОтправки = -1 Тогда
ВызватьИсключение "Ошибка отправки Protobuf: "
+ Компонента.ПрочитатьСообщениеОбОшибке();
КонецЕсли;
// *** КРИТИЧНО: Сохраняем offset в транзакцию ***
// Формируем JSON с информацией об офсете (+1, т.к. фиксируем следующий офсет для чтения)
ОфсетыJSON = СтрШаблон(
"[{""topic"": ""%1"", ""partition"": %2, ""offset"": %3}]",
ТопикСообщения,
ПартицияСообщения,
ОфсетСообщения + 1
);
// Это гарантирует, что offset будет зафиксирован только вместе с отправленным сообщением
Компонента.ОтправитьОфсетыВТранзакцию(ОфсетыJSON, ГруппаПотребителей);
// *** ФИКСИРУЕМ ТРАНЗАКЦИЮ KAFKA ***
// Атомарно фиксируются: отправленное сообщение + offset прочитанного сообщения
Компонента.ЗафиксироватьТранзакцию();
// Статистика (только после успешной фиксации)
КоличествоОбработано = КоличествоОбработано + 1;
Если СтатусОбработки.Статус = "CONFIRMED" Тогда
КоличествоПодтверждено = КоличествоПодтверждено + 1;
ИначеЕсли СтатусОбработки.Статус = "REJECTED" Тогда
КоличествоОтклонено = КоличествоОтклонено + 1;
КонецЕсли;
// Логируем каждые 100 заказов
Если КоличествоОбработано % 100 = 0 Тогда
Сообщить(СтрШаблон(
"Обработано заказов: %1 (подтверждено: %2, отклонено: %3)",
КоличествоОбработано,
КоличествоПодтверждено,
КоличествоОтклонено
));
КонецЕсли;
Исключение
// *** ОТКАТЫВАЕМ ТРАНЗАКЦИЮ KAFKA при любой ошибке ***
// Сообщение будет прочитано заново при следующей итерации
Компонента.ОтменитьТранзакцию();
ВызватьИсключение;
КонецПопытки;
Иначе
// Таймаут - сообщений нет, небольшая пауза
// В production здесь можно проверять условие остановки обработки
КонецЕсли;
Исключение
// Логируем ошибку, но продолжаем обработку
ТекстОшибки = ОписаниеОшибки();
Сообщить("ОШИБКА: " + ТекстОшибки, СтатусСообщения.Важное);
// В production здесь можно отправлять ошибку в систему мониторинга
// или в Dead Letter Queue
КонецПопытки;
КонецЦикла;
КонецПроцедуры
Функция ОбработатьЗаказВ1С(ДанныеЗаказа)
// Структура для возврата статуса
Статус = Новый Структура;
Статус.Вставить("ИдентификаторЗаказа", ДанныеЗаказа["order_id"][0]);
Статус.Вставить("ВремяОбработки", ТекущаяУниверсальнаяДатаВМиллисекундах());
Статус.Вставить("ОбработалСистема", "1C-ERP-Instance-01");
Статус.Вставить("Резервирования", Новый Массив);
Статус.Вставить("СуммаРезерва", 0);
Попытка
// Начинаем транзакцию в 1С
НачатьТранзакцию();
Попытка
// Проверяем остатки и резервируем товары
МассивПозиций = ДанныеЗаказа["items"][0]; // AVRO возвращает массивы в формате columnar
ВсеТоварыДоступны = Истина;
ОбщаяСуммаРезерва = 0;
Для Индекс = 0 По МассивПозиций.Количество() - 1 Цикл
ИдентификаторТовара = ДанныеЗаказа["items"][0]["product_id"][Индекс];
Количество = ДанныеЗаказа["items"][0]["quantity"][Индекс];
Цена = ДанныеЗаказа["items"][0]["price"][Индекс];
// Находим номенклатуру
Номенклатура = Справочники.Номенклатура.ПолучитьСсылку(
Новый УникальныйИдентификатор(ИдентификаторТовара)
);
// Проверяем остаток (упрощённая логика)
Остаток = ПолучитьОстатокНоменклатуры(Номенклатура);
Если Остаток >= Количество Тогда
// Резервируем товар
РезервТовар = РезервироватьТовар(Номенклатура, Количество);
// Добавляем в массив резервирований
СтруктураРезерв = Новый Структура;
СтруктураРезерв.Вставить("product_id", ИдентификаторТовара);
СтруктураРезерв.Вставить("warehouse_id", Строка(РезервТовар.Склад.УникальныйИдентификатор()));
СтруктураРезерв.Вставить("quantity", Количество);
СтруктураРезерв.Вставить("reserved_amount", Количество * Цена);
Статус.Резервирования.Добавить(СтруктураРезерв);
ОбщаяСуммаРезерва = ОбщаяСуммаРезерва + (Количество * Цена);
Иначе
// Недостаточно товара
ВсеТоварыДоступны = Ложь;
Статус.Вставить("ПричинаОтказа", СтрШаблон(
"Недостаточно товара %1. Требуется: %2, доступно: %3",
Номенклатура,
Количество,
Остаток
));
Прервать;
КонецЕсли;
КонецЦикла;
Если ВсеТоварыДоступны Тогда
// Фиксируем транзакцию
ЗафиксироватьТранзакцию();
Статус.Вставить("Статус", "CONFIRMED");
Статус.Вставить("СуммаРезерва", ОбщаяСуммаРезерва);
Статус.Вставить("ПричинаОтказа", "");
Сообщить(СтрШаблон(
"Заказ %1 подтверждён, зарезервировано на сумму: %2 руб.",
Статус.ИдентификаторЗаказа,
Формат(ОбщаяСуммаРезерва, "ЧЦ=15; ЧДЦ=2")
));
Иначе
// Откатываем транзакцию
ОтменитьТранзакцию();
Статус.Вставить("Статус", "REJECTED");
Статус.Резервирования.Очистить();
КонецЕсли;
Исключение
ОтменитьТранзакцию();
ВызватьИсключение;
КонецПопытки;
Исключение
Статус.Вставить("Статус", "REJECTED");
Статус.Вставить("ПричинаОтказа", "Внутренняя ошибка обработки: " + КраткоеПредставлениеОшибки(ИнформацияОбОшибке()));
Статус.Резервирования.Очистить();
Сообщить(СтрШаблон(
"Заказ %1 отклонён: %2",
Статус.ИдентификаторЗаказа,
Статус.ПричинаОтказа
), СтатусСообщения.Важное);
КонецПопытки;
Возврат Статус;
КонецФункции
Функция СформироватьСтатусJSON(СтатусОбработки)
// Формируем JSON для Protobuf
СтруктураJSON = Новый Структура;
СтруктураJSON.Вставить("order_id", СтатусОбработки.ИдентификаторЗаказа);
СтруктураJSON.Вставить("status", СтатусОбработки.Статус);
СтруктураJSON.Вставить("processed_at", СтатусОбработки.ВремяОбработки);
СтруктураJSON.Вставить("processed_by", СтатусОбработки.ОбработалСистема);
СтруктураJSON.Вставить("reservations", СтатусОбработки.Резервирования);
СтруктураJSON.Вставить("rejection_reason", СтатусОбработки.Получить("ПричинаОтказа"));
СтруктураJSON.Вставить("total_reserved_amount", СтатусОбработки.СуммаРезерва);
ЗаписьJSON = Новый ЗаписьJSON;
ЗаписьJSON.УстановитьСтроку();
ЗаписатьJSON(ЗаписьJSON, СтруктураJSON);
СтрокаJSON = ЗаписьJSON.Закрыть();
Возврат СтрокаJSON;
КонецФункции
Функция ТекущаяУниверсальнаяДатаВМиллисекундах()
Возврат (ТекущаяУниверсальнаяДата() - Дата(1970, 1, 1)) * 1000;
КонецФункции
// Упрощённые вспомогательные функции (в реальности сложнее)
Функция ПолучитьОстатокНоменклатуры(Номенклатура)
// Здесь запрос к регистру остатков
Возврат 100; // Для примера
КонецФункции
Функция РезервироватьТовар(Номенклатура, Количество)
// Здесь создание записи регистра резервов
Результат = Новый Структура;
// Заполнение результата
Возврат Результат;
КонецФункции
Результат работы примера с Exactly-Once:
- READ (AVRO) — получаем заказы от Order Service (AVRO обеспечивает самодокументируемость и гибкую эволюцию схемы)
- PROCESS (1С) — проверяем остатки, резервируем товары в транзакции БД 1С
- WRITE (Protobuf) — отправляем компактный статус обработки для микросервисов в транзакции Kafka
- COMMIT — фиксируем транзакцию Kafka вместе с offset — гарантия Exactly-Once!
- Аналитика читает — Analytics System читает данные из обоих топиков для построения отчётов и дашбордов
Ключевые механизмы Exactly-Once:
Преимущества такого подхода:
- Правильная архитектура — каждый формат используется по назначению: AVRO для событий с эволюцией схемы, Protobuf для высокопроизводительных статусов
- 1С как центральный процессор — обрабатывает критичную бизнес-логику (остатки, резервирование) между разнородными микросервисами
- Производительность — обработка тысяч заказов в минуту благодаря бинарным форматам и эффективной сериализации
- Надёжность — транзакции в 1С + Kafka consumer groups гарантируют целостность данных и at-least-once доставку
- Масштабируемость — можно запустить несколько экземпляров 1С в одной consumer group для параллельной обработки
Пример 4: Гибридный подход для отладки
Один из умных приемов — использовать JSON для разработки и отладки, а AVRO/Protobuf для production:
Функция ОтправитьСообщениеУниверсально(Данные, Топик, ФорматСообщения = "JSON")
Если ФорматСообщения = "AVRO" Тогда
// Преобразуем в AVRO
Компонента.ПреобразоватьВФорматAVRO(Данные, "sales_document");
Результат = Компонента.ОтправитьСообщениеAVRO(Топик);
ИначеЕсли ФорматСообщения = "PROTOBUF" Тогда
// Преобразуем в Protobuf
Компонента.ПреобразоватьВФорматProtobuf(Данные, "StockUpdate");
Результат = Компонента.ОтправитьСообщениеProtobuf(Топик);
Иначе
// Отправляем как обычный JSON
Результат = Компонента.ОтправитьСообщение(Данные, Топик);
КонецЕсли;
Возврат Результат;
КонецФункции
// Использование:
// Разработка: ОтправитьСообщениеУниверсально(Данные, "test.topic", "JSON");
// Production: ОтправитьСообщениеУниверсально(Данные, "prod.topic", "PROTOBUF");
Заключение
Поддержка AVRO и Protobuf в компоненте Simple Kafka Connector — это не просто добавление "еще двух форматов". Это качественный скачок в возможностях 1С как платформы интеграции.
Теперь 1С может:
- Говорить на одном языке с современными микросервисами (без промежуточных конвертеров)
- Обрабатывать большие объемы данных благодаря 10-кратному ускорению и 70% экономии трафика
- Участвовать в событийно-ориентированных архитектурах наравне с сервисами на Go, Java, Python
Это открывает дорогу к новым архитектурным паттернам: Event Sourcing, CQRS, Real-time analytics — вещам, которые раньше казались недостижимыми в мире 1С.
Практические преимущества
Для бизнеса:
- Снижение стоимости инфраструктуры — меньше трафика = меньше расходы на облачные сервисы
- Ускорение интеграций — данные обрабатываются в 5-10 раз быстрее
- Масштабируемость — возможность обрабатывать миллионы событий в день
Для разработчиков:
- Типобезопасность — схемы гарантируют корректность данных
- Версионирование — безболезненная эволюция форматов данных
- Совместимость — интеграция с любыми современными системами
С чего начать
- Изучите вашу архитектуру — определите узкие места в интеграциях
- Выберите формат — AVRO для аналитики и гибкости, Protobuf для производительности
- Начните с пилота — выберите один поток данных и внедрите бинарный формат
- Измеряйте результаты — сравните производительность до и после
Компонента доступна на GitHub с полной документацией и примерами. Начните с малого, и вы увидите, как 1С из "тяжеловесной системы учета" превращается в быстрый и гибкий компонент современной микросервисной архитектуры.
Полезные ссылки:
Документация компоненты:
Внешние ресурсы:
