Использовать последнее значение из темы для каждого ключа - PullRequest
Новая
       17

Использовать последнее значение из темы для каждого ключа

0 голосов
/ 27 апреля 2018

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

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

Необработанное решение, которое было реализовано, было таким: производитель отправляет только ключ, когда каждое сообщение записывается, а фактическое значение записывается в базу данных, которая передается потребителю. Потребитель читает каждый ключ из очереди и значение из базы данных. Здесь цель всегда читать последнее значение достигается тем, что производитель перезаписывает значение в базе данных, поэтому потребитель, который фактически читает данный ключ, будет фактически потреблять последнее значение. Но у этого решения есть некоторые недостатки из-за большого количества операций чтения и обновления (медленные, условия гонки и т. Д.)

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

1 Ответ

0 голосов
/ 17 мая 2018

Ниже помог код 

KStreamBuilder builder = new KStreamBuilder();
KTable<String, String> dataTable = builder.table("input-topic");
dataTable.toStream().foreach((key, message) -> client.post(message));
KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();

Что делает это возможным на практике, так это сжатие входящего потока в памяти (подробности объяснены здесь ). Мы могли бы контролировать давление, используя параметры cache.max.bytes.buffering и commit.interval.ms

...