Можно ли последовательно обрабатывать сообщения с помощью API-интерфейса Kafka Streams? - PullRequest
Новая
       6

Можно ли последовательно обрабатывать сообщения с помощью API-интерфейса Kafka Streams?

0 голосов
/ 25 октября 2018

У нас есть несколько сообщений, последовательность которых нам нужна.Мы решили отправить все сообщения из определенного источника в раздел, чтобы сохранить последовательность сообщений ( несколько источников могут создавать один и тот же раздел, но источник не может создавать несколько разделов ), и мы будембыть в состоянии идентифицировать каждый источник по их ключу.

Теперь нам нужно использовать эти сообщения и выполнить некоторую обработку.Мы делаем несколько независимых операций над сообщениями, которые мы потребили (например, сохраняя их в базе данных, пересылая их и т. Д.).Теперь я задаюсь вопросом, использовать ли для этого Kafka Streams API или Consumer API.

Примечание: у меня не может быть большого количества тем (например, я не могу создать тему для каждого источника, так как источники будутбыть многочисленным).Хотя я могу группировать сообщения по ключу, чтобы определить источник, для использования потоков мне нужен порядок сообщений для ключа.

Вариант использования: я хочу зафиксировать эти сообщения в базе данных , чтобы и я хочу переслать эти сообщения в порядке .

Так, как я могу обработать сообщения в порядке, используя Streams API?

1 Ответ

0 голосов
/ 25 октября 2018

Мы решили отправить все сообщения из определенного источника в раздел

Кафка гарантирует порядок в пределах одного раздела, в порядке их поступления в брокер, а не каким-либо другимполе или по времени.Все клиенты Kafka (Consumer, Streams, сторонние библиотеки) уважают этот факт.

В общем, хотя, если вы включите метку времени сообщения как часть события вставки вашей базы данных, то можно будет сгруппировать по ключу, порядок по отметке времени.Однако это будет зависеть от базы данных.

Если вы уже делите разделы по известному ключу источника, почему бы вам не выполнить фильтрацию по этому в приложении потоков?В противном случае вам придется использовать Consumer API, поскольку он позволяет назначать определенные разделы (возможно, API-процессор, но не использовал его)

...