流消息到多个主题

本文介绍了流消息到多个主题的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我有一个主主题和多个谓词，每个谓词都有一个与之关联的输出主题.我想将每个记录发送到谓词解析为true的所有主题.我正在使用Luwak来测试哪个谓词满足记录(要使用此库，您可以评估带有谓词列表的文档，并告诉您匹配的谓词-即，我只调用一次即可获得满意的谓词列表).

我正在尝试为此使用Kafka Streams，但似乎在KStream上没有合适的方法(KStream#branch仅将记录路由到单个主题).

一种可能的方法如下:

Stream from master Map the values into a format with the original content and the list of matching predicates Stream to an intermediate with-matches topic For each predicate/output topic Stream from intermediate with-matches topic Filter "does list of matches predicates contain predicate ID" Map the values to just the original content Stream to corresponding output topic

这种中间主题似乎笨拙".有更好的建议吗?

我正在使用:

• Kafka v0.10.1.1
• Luwak v1.4.0

解决方案

您可以简单地将多个过滤器并行应用于同一KStream实例:

KStream stream = ... stream.filter(new MyPredicate1()).to("output-topic-1"); stream.filter(new MyPredicate2()).to("output-topic-2"); stream.filter(new MyPredicate3()).to("output-topic-3"); // ... as as many as you need

每条记录将发送到每个谓词一次-从概念上讲，这是对所有筛选器的广播，但是不会物理复制记录，因此没有内存开销.

I have a single master topic and multiple predicates each of which has an output topic associated with it. I want to send each record to ALL topics that whose predicate resolves to true. I am using Luwak to test which predicates a record satisfies (to use this library you evaluate a document with a list of predicates and it tells you which ones matched - i.e. I only call it once to get the list of satisfied predicates).

I am trying to use Kafka Streams for this but there doesn't seem to be the appropriate method on KStream (KStream#branch only routes a record to a single topic).

One possible approach is as follows:

Stream from master Map the values into a format with the original content and the list of matching predicates Stream to an intermediate with-matches topic For each predicate/output topic Stream from intermediate with-matches topic Filter "does list of matches predicates contain predicate ID" Map the values to just the original content Stream to corresponding output topic

Such an intermediate topic seems "clunky" though. Any better suggestions?

I am using:

• Kafka v0.10.1.1
• Luwak v1.4.0

解决方案

You can simple apply multiple filters in parallel to the same KStream instance:

KStream stream = ... stream.filter(new MyPredicate1()).to("output-topic-1"); stream.filter(new MyPredicate2()).to("output-topic-2"); stream.filter(new MyPredicate3()).to("output-topic-3"); // ... as as many as you need

Each record will be sent to each predicate once -- it's conceptually a broadcast to all filters, but records will not be physically replicated, so there is no memory overhead.