我正在使用帶有 topicPattern 的 Spring @KafkaListener。如果在此應用程式的運行期間我創建了一個與該模式匹配的新主題并開始發布到該主題，則偵聽器應用程式將忽略這些訊息。換句話說，它只在啟動時提取與模式匹配的所有主題并聽取這些主題。
“重繪 ”它的最簡單方法是什么？謝謝！
uj5u.com熱心網友回復：
默認情況下，根據https://kafka.apache.org/documentation/#consumerconfigs_metadata.max.age.ms的設定，會在 5 分鐘內（默認）拾取新主題
即使我們沒有看到任何磁區領導層更改以主動發現任何新代理或磁區，我們也強制重繪 元資料的時間段（以毫秒為單位）。
您可以減少它以加快速度，但會增加流量。
編輯
這表明它按預期作業......
@SpringBootApplication
public class So71386069Application {
private static final Logger log = LoggerFactory.getLogger(So71386069Application.class);
public static void main(String[] args) {
SpringApplication.run(So71386069Application.class, args);
}
@KafkaListener(id = "so71386069", topicPattern = "so71386069.*",
properties = "metadata.max.age.ms:60000")
void listen(String in) {
System.out.println(in);
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("so71386069").partitions(1).replicas(1).build();
}
@Bean
ApplicationRunner runner(KafkaAdmin admin) {
return args -> {
try (AdminClient client = AdminClient.create(admin.getConfigurationProperties())) {
IntStream.range(0, 10).forEach(i -> {
try {
Thread.sleep(30_000);
String topic = "so71386069-" i;
log.info("Creating {}", topic);
client.createTopics(Collections.singleton(
TopicBuilder.name(topic).partitions(1).replicas(1).build())).all().get();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
catch (ExecutionException e) {
e.printStackTrace();
}
});
}
};
}
}
2022-03-07 15:41:07.131 INFO 33630 --- [o71386069-0-C-1] o.s.k.l.KafkaMessageListenerContainer
: so71386069: partitions assigned: [so71386069-0]
2022-03-07 15:41:34.007 INFO 33630 --- [ main] com.example.demo.So71386069Application
: Creating so71386069-0
2022-03-07 15:42:04.193 INFO 33630 --- [ main] com.example.demo.So71386069Application
: Creating so71386069-1
...
2022-03-07 15:42:07.590 INFO 33630 --- [o71386069-0-C-1] o.s.k.l.KafkaMessageListenerContainer
: so71386069: partitions revoked: [so71386069-0]
...
2022-03-07 15:42:07.599 INFO 33630 --- [o71386069-0-C-1] o.s.k.l.KafkaMessageListenerContainer
: so71386069: partitions assigned: [so71386069-0, so71386069-1-0, so71386069-0-0]
2022-03-07 15:42:34.378 INFO 33630 --- [ main] com.example.demo.So71386069Application
: Creating so71386069-2
2022-03-07 15:43:04.554 INFO 33630 --- [ main] com.example.demo.So71386069Application
: Creating so71386069-3
...
2022-03-07 15:43:08.403 INFO 33630 --- [o71386069-0-C-1] o.s.k.l.KafkaMessageListenerContainer
: so71386069: partitions revoked: [so71386069-0, so71386069-1-0, so71386069-0-0]
...
2022-03-07 15:43:08.411 INFO 33630 --- [o71386069-0-C-1] o.s.k.l.KafkaMessageListenerContainer
: so71386069: partitions assigned: [so71386069-0, so71386069-3-0, so71386069-2-0, so71386069-1-0, so71386069-0-0]
...
uj5u.com熱心網友回復：
我認為這就是設計的方式。Kafka 客戶端始終必須訂閱主題才能獲取訊息。
在這種情況下，在啟動時，Kafka 客戶端/消費者會在啟動時訂閱匹配模式的主題，這就是它所進行的。
但這確實是一個有趣的問題。最簡單最簡單的答案是“重新啟動客戶端/消費者”。但是，將密切關注其他人的答案以了解任何想法。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/440034.html
標籤：春天 阿帕奇卡夫卡 运行 卡夫卡消费者API 春天卡夫卡