Skip to content

Commit ad34782

Browse files
Segate-ekbSegate-ekb
committed
consumers impl
1 parent 409c335 commit ad34782

29 files changed

Lines changed: 2487 additions & 7 deletions

File tree

‎doc/specs/kafka-consumer-manager.md‎

Lines changed: 202 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,202 @@
1+
# Kafka Consumer Manager для входящих потоков
2+
3+
## Статус: В разработке
4+
5+
## Обзор
6+
7+
Реализация менеджера входящих потоков с новым справочником `инт_Подписки` для настройки консьюмеров Kafka. Менеджер запускается раз в 30 секунд, группирует потоки по режиму обработки:
8+
9+
- **Асинхронные потоки** → единая группа консьюмеров, сообщения записываются в `инт_ОчередьВходящихСообщений`
10+
- **Синхронные потоки** → отдельные ФЗ на каждую подписку, немедленная обработка с отправкой в DLQ при ошибках
11+
12+
## Архитектура
13+
14+
```
15+
┌─────────────────────────────────────────────────────────────────────┐
16+
│ РЗ: инт_МенеджерВходящихПотоков │
17+
│ (каждые 30 сек) │
18+
└─────────────────────────────────────────────────────────────────────┘
19+
│
20+
┌───────────────┴───────────────┐
21+
▼ ▼
22+
┌───────────────────────┐ ┌───────────────────────┐
23+
│ Асинхронные потоки │ │ Синхронные потоки │
24+
│ (АсинхроннаяОбработка│ │ (АсинхроннаяОбработка│
25+
│ = Истина) │ │ = Ложь) │
26+
└───────────────────────┘ └───────────────────────┘
27+
│ │
28+
▼ ▼
29+
┌───────────────────────┐ ┌───────────────────────┐
30+
│ 1 ФЗ на все потоки │ │ 1 ФЗ на каждую │
31+
│ ConsumerGroup: │ │ подписку │
32+
│ {GUID_ИБ}_async │ │ ConsumerGroup: │
33+
│ │ │ из настроек подписки │
34+
└───────────────────────┘ └───────────────────────┘
35+
│ │
36+
▼ ▼
37+
┌───────────────────────┐ ┌───────────────────────┐
38+
│ Сообщение → │ │ Сообщение → │
39+
│ инт_ОчередьВходящих │ │ Немедленная обработка│
40+
│ Сообщений │ │ Ошибка → DLQ топик │
41+
└───────────────────────┘ └───────────────────────┘
42+
```
43+
44+
## Компоненты
45+
46+
### 1. Определяемый тип `инт_Подписка`
47+
48+
Полиморфный тип для подписчиков, требующих активного получения сообщений (консьюмеры).
49+
50+
```xml
51+
<Type>
52+
<v8:Type>cfg:CatalogRef.инт_ПодписчикиKafka</v8:Type>
53+
<v8:Type>cfg:CatalogRef.инт_ПодписчикиRabbitMQ</v8:Type>
54+
</Type>
55+
```
56+
57+
### 2. Доработка справочника `инт_ПодписчикиKafka`
58+
59+
Добавить новый реквизит в область "Параметры консьюмера":
60+
61+
| Реквизит | Тип | Описание |
62+
|----------|-----|----------|
63+
| `ИмяDLQТопика` | String(255) | Топик для "мёртвых" сообщений (Dead Letter Queue) |
64+
65+
**Consumer Group ID** генерируется автоматически: `{GUID_ИБ}_{Код подписчика}`
66+
67+
### 3. Табличная часть `ПодпискиНаВходящиеСообщения` в `инт_ПотокиДанных`
68+
69+
| Колонка | Тип | Описание |
70+
|---------|-----|----------|
71+
| `Подписка` | DefinedType.инт_Подписка | Ссылка на подписчика (Kafka/RabbitMQ) |
72+
73+
**Условная видимость**: ТЧ видима только когда `НаправлениеПотока = Входящий`
74+
75+
### 4. Константа `инт_ТаймаутОжиданияКонсьюмера`
76+
77+
- **Тип**: Number
78+
- **Значение по умолчанию**: 5000 (мс)
79+
- **Назначение**: Таймаут ожидания сообщений в методе `ПрочитатьСообщение()`
80+
81+
### 5. Регламентное задание `инт_МенеджерВходящихПотоков`
82+
83+
- **Расписание**: каждые 30 секунд
84+
- **Метод**: `инт_МенеджерВходящихПотоков.ЗапуститьМенеджер()`
85+
- **Назначение**: Мониторинг и запуск фоновых заданий консьюмеров
86+
87+
### 6. Общий модуль `инт_МенеджерВходящихПотоков`
88+
89+
#### Экспортные методы
90+
91+
```bsl
92+
// Точка входа регламентного задания
93+
Процедура ЗапуститьМенеджер() Экспорт
94+
95+
// Запуск консьюмера для асинхронных потоков (одно ФЗ на все)
96+
Процедура ЗапуститьАсинхронныйКонсьюмер(МассивПодписок) Экспорт
97+
98+
// Запуск консьюмера для синхронного потока (отдельное ФЗ)
99+
Процедура ЗапуститьСинхронныйКонсьюмер(Подписка, ПотокДанных) Экспорт
100+
```
101+
102+
#### Алгоритм `ЗапуститьМенеджер()`
103+
104+
1. Получить активные входящие потоки с заполненными подписками
105+
2. Разделить на 2 группы по флагу `АсинхроннаяОбработка`
106+
3. Для асинхронной группы:
107+
- Проверить, запущено ли ФЗ `инт_КонсьюмерАсинхронныйГрупповой`
108+
- Если нет или завершено — запустить новое
109+
4. Для синхронной группы:
110+
- По каждой подписке проверить наличие активного ФЗ
111+
- Если нет — запустить `инт_КонсьюмерСинхронный`
112+
113+
### 7. Консьюмер-логика в `инт_РаботаСКафка`
114+
115+
#### Новые методы
116+
117+
```bsl
118+
// Создание и инициализация консьюмера
119+
// Возврат: Компонента (внешняя компонента simpleKafka1C)
120+
Функция СоздатьКонсьюмер(Брокеры, ГруппаКонсьюмеров, МассивТопиков) Экспорт
121+
122+
// Цикл чтения сообщений для асинхронного режима
123+
// Сообщения записываются в инт_ОчередьВходящихСообщений
124+
Процедура ЧитатьСообщенияАсинхронно(Компонента, СоответствиеТопикПоток) Экспорт
125+
126+
// Цикл чтения сообщений для синхронного режима
127+
// Сообщения обрабатываются сразу, ошибки → DLQ
128+
Процедура ЧитатьСообщенияСинхронно(Компонента, Подписка, ПотокДанных) Экспорт
129+
130+
// Отправка сообщения в Dead Letter Queue
131+
Процедура ОтправитьВDLQ(Компонента, DLQТопик, ИсходноеСообщение, ОшибкаОбработки, КоличествоПопыток) Экспорт
132+
```
133+
134+
### 8. Регистр сведений `инт_АктивныеКонсьюмеры`
135+
136+
Мониторинг состояния фоновых заданий консьюмеров.
137+
138+
| Тип | Имя | Тип данных | Описание |
139+
|-----|-----|------------|----------|
140+
| Измерение | `ИдентификаторЗадания` | UUID | ID фонового задания |
141+
| Ресурс | `ПотокДанных` | CatalogRef.инт_ПотокиДанных | Связанный поток (для sync) |
142+
| Ресурс | `Подписка` | DefinedType.инт_Подписка | Связанный подписчик (Kafka/RabbitMQ) |
143+
| Ресурс | `РежимОбработки` | String(20) | "Асинхронный" / "Синхронный" |
144+
| Ресурс | `ВремяЗапуска` | DateTime | Время старта ФЗ |
145+
| Ресурс | `ПоследняяАктивность` | DateTime | Время последнего heartbeat |
146+
147+
## Формат DLQ-сообщения
148+
149+
```json
150+
{
151+
"original_topic": "orders.incoming",
152+
"original_partition": 0,
153+
"original_offset": 12345,
154+
"original_payload": "{...исходные данные...}",
155+
"error_message": "Ошибка валидации: поле 'amount' обязательно",
156+
"retry_count": 3,
157+
"failed_at": "2026-02-02T14:30:00Z"
158+
}
159+
```
160+
161+
## Consumer Group Strategy
162+
163+
Consumer Group ID генерируется автоматически:
164+
165+
- **Асинхронные потоки**: `{GUID_ИБ}_async_incoming` — единая группа для всех async-потоков
166+
- **Синхронные потоки**: `{GUID_ИБ}_{Код подписчика}` — отдельная группа для каждого подписчика
167+
168+
GUID информационной базы получается через `ПараметрыСеанса.ИдентификаторИБ` или `РегистрыСведений.инт_ПараметрыИнтеграции`.
169+
170+
## Цикл чтения сообщений
171+
172+
```bsl
173+
Пока РазрешеноСлушать Цикл
174+
РезультатЧтения = Компонента.ПрочитатьСообщение();
175+
176+
Если НЕ РезультатЧтения Тогда
177+
// Проверка флага Активен подписки
178+
РазрешеноСлушать = ОбщегоНазначения.ЗначениеРеквизитаОбъекта(Подписка, "Активен");
179+
Если Не РазрешеноСлушать Тогда
180+
Прервать;
181+
Иначе
182+
Продолжить;
183+
КонецЕсли;
184+
КонецЕсли;
185+
186+
ДанныеСообщения = Компонента.ПолучитьДанныеСообщения();
187+
ТопикСообщения = Компонента.ПолучитьТопикСообщения();
188+
СмещениеСообщения = Компонента.ПолучитьСмещениеСообщения();
189+
190+
// Обработка в зависимости от режима...
191+
192+
// Фиксация смещения
193+
НовоеСмещение = Число(СмещениеСообщения) + 1;
194+
Компонента.ЗафиксироватьСмещение(ТопикСообщения, НовоеСмещение, РазделСообщения);
195+
КонецЦикла;
196+
```
197+
198+
## Ссылки
199+
200+
- [Kafka в 1С: Exactly-Once и DLQ](https://infostart.ru/1c/articles/2575918/)
201+
- [Simple Kafka Connector 1C](https://github.com/NuclearAPK/Simple-Kafka_Adapter)
202+
- Пример консьюмера: [examples/consumer.bsl](../../examples/consumer.bsl)

‎docs/concepts/index.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -138,5 +138,7 @@ outline: deep
138138

139139
- [Потоки данных](/concepts/data-flows) — подробнее о конфигурации потоков
140140
- [Подписчики](/concepts/subscribers) — типы и настройка подписчиков
141+
- [Kafka-консьюмеры](/concepts/kafka-consumers) — входящие потоки из Kafka
141142
- [Валидация](/concepts/validation) — работа с OpenAPI-схемами
142-
- [Многопоточность](/concepts/multithreading) — параллельная обработка- [Мониторинг и метрики](/concepts/monitoring) — Prometheus-метрики и алерты
143+
- [Многопоточность](/concepts/multithreading) — параллельная обработка
144+
- [Мониторинг и метрики](/concepts/monitoring) — Prometheus-метрики и алерты

0 commit comments

Comments
 (0)