我有一个主题叫mytopic。这个主题有一个生产者和两个消费者。我需要做的是根据消息的前缀过滤生产者生成的消息。例如，如果消息以“a”前缀开头，则只有第一个使用者必须接受它。如果它以“b”前缀开头，则只有第二个使用者必须使用它。我做了很多搜索，我发现的是过滤来自某个主题的消息，然后在过滤后将它们发送到不同的主题。但是如上所述，我需要对一个主题进行过滤。在Kafka我怎么能做到？
7vux5j2d1#
允许两个使用者都使用所有数据，一旦获得记录，就使用java流和特定于使用者的过滤逻辑对它们进行过滤。简言之，我的意思是只按原样获取数据并使用java代码过滤它们，而不是在kafka级别进行过滤。更新：如果要在kafka级别进行过滤，可以使用分区，在向kafka主题发送消息时，将前缀为“a”的消息发送到分区1，将前缀为“b”的消息发送到分区2。现在，在消费时，只需在各自的消费者中消费特定的分区。
hof1towb2#
这很简单，不需要写回不同的主题。“2个使用者”是指2个使用者组或1个使用者组中的2个使用者线程？我两个都说。如果是1个使用者组中的2个使用者线程，则可以使用消息“key”字段。Kafka将相同的“密钥”消息发送到相同的“分区”。例如，一个消息前缀“a”带有关键字字段“a”，b消息前缀“b”带有关键字字段“b”，那么kafka将把a消息发送到“partition-1”，把b消息发送到“partition-2”。使用者线程a可以订阅指定的'mytopic-partition-1'，线程b可以使用'org.apache.kafka.clients.consumer.kafkaconsumer'类中的'assign'方法订阅'mytopic-partition-2'。如果是2个消费群体，只需订阅主题并在代码中筛选即可。如果不满意，请使用上述相同的方法。诀窍是将特定的前缀消息发送到特定的“分区”。如果你真的想过滤，也许你可以使用Kafka连接插件。
2条答案
按热度按时间7vux5j2d1#
允许两个使用者都使用所有数据，一旦获得记录，就使用java流和特定于使用者的过滤逻辑对它们进行过滤。
简言之，我的意思是只按原样获取数据并使用java代码过滤它们，而不是在kafka级别进行过滤。
更新：
如果要在kafka级别进行过滤，可以使用分区，在向kafka主题发送消息时，将前缀为“a”的消息发送到分区1，将前缀为“b”的消息发送到分区2。
现在，在消费时，只需在各自的消费者中消费特定的分区。
hof1towb2#
这很简单，不需要写回不同的主题。
“2个使用者”是指2个使用者组或1个使用者组中的2个使用者线程？
我两个都说。
如果是1个使用者组中的2个使用者线程，则可以使用消息“key”字段。
Kafka将相同的“密钥”消息发送到相同的“分区”。
例如，一个消息前缀“a”带有关键字字段“a”，b消息前缀“b”带有关键字字段“b”，那么kafka将把a消息发送到“partition-1”，把b消息发送到“partition-2”。使用者线程a可以订阅指定的'mytopic-partition-1'，线程b可以使用'org.apache.kafka.clients.consumer.kafkaconsumer'类中的'assign'方法订阅'mytopic-partition-2'。
如果是2个消费群体，只需订阅主题并在代码中筛选即可。如果不满意，请使用上述相同的方法。
诀窍是将特定的前缀消息发送到特定的“分区”。
如果你真的想过滤，也许你可以使用Kafka连接插件。