如何通过springboot Kafka中的kafkalistener服务根据传入的主题模式自动创建Kafka主题?
How to AutoCreate Kafka Topic according to incoming topic pattern through kafkalistener service in springboot Kafka?
我将订阅 kafka 主题模式,例如“topic.*”。我的目标是为我收听的每个 kafka 主题创建死信队列。
例如,当我收听名为“topic.1”的主题时,我想自动创建名为“topic.1_deadletter”的死信队列。
到目前为止我尝试做的如下:
我的消费者:
@Component
@Slf4j
public class LibraryEventsConsumer {
@Autowired
LibraryEventConsumerConfig libraryEventConsumerConfig;
@KafkaListener(topicPattern = "kafka.*")
public void onMessage(String consumerRecord, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws Exception{
log.info("ConsumerRecord : {}", consumerRecord);
String deadlettertopic = String.format("%s_deadletter",topic);
System.out.println(deadlettertopic);
System.out.println(KafkaHeaders.RECEIVED_TOPIC);
libraryEventConsumerConfig.getTopic(topic);`
我在这里使用方法 getTopic 尝试自动创建 kafka 主题。在下面你可以看到 libraryEventConsumer class:
@Configuration
@EnableKafka
public class LibraryEventConsumerConfig {
@Bean
public void getTopic(String topic){
NewTopic deadlettertopic = TopicBuilder.name(String.format("%s_deadletter",topic))
.partitions(1)
.replicas(1)
.build();
}
}
不幸的是,该方法无效,我收到以下错误消息:
Parameter 0 of method getTopic in com.kafkalibrary.Config.LibraryEventConsumerConfig required a bean of type 'java.lang.String' that could not be found.
知道如何进行吗?
解决方案示例代码:
对于那些正在寻找相同目标的人,这是我的示例代码:感谢 Gary Russell 的启发。
private static void createTopic(String topicName, int numPartitions) throws Exception {
Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:5052,localhost:5053,localhost:5054");
AdminClient admin = AdminClient.create(config);
//checking if topic already exists
boolean alreadyExists = admin.listTopics().names().get().stream()
.anyMatch(existingTopicName -> existingTopicName.equals(topicName));
if (alreadyExists) {
System.out.printf("topic already exits: %s%n", topicName);
} else {
//creating new topic
System.out.printf("creating topic: %s%n", topicName);
NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) 1);
admin.createTopics(Collections.singleton(newTopic)).all().get();
}
添加再平衡侦听器,或扩展 AbstractConsumerSeekAware
(或仅实施 ConsumerSeekAware
)。
public class LibraryEventsConsumer extends AbstractConsumerSeekAware {
然后,在 onPartitionsAssigned()
中使用 AdminClient
检查 DLT 主题是否存在,如果不存在,则创建它。
我将订阅 kafka 主题模式,例如“topic.*”。我的目标是为我收听的每个 kafka 主题创建死信队列。
例如,当我收听名为“topic.1”的主题时,我想自动创建名为“topic.1_deadletter”的死信队列。
到目前为止我尝试做的如下:
我的消费者:
@Component
@Slf4j
public class LibraryEventsConsumer {
@Autowired
LibraryEventConsumerConfig libraryEventConsumerConfig;
@KafkaListener(topicPattern = "kafka.*")
public void onMessage(String consumerRecord, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws Exception{
log.info("ConsumerRecord : {}", consumerRecord);
String deadlettertopic = String.format("%s_deadletter",topic);
System.out.println(deadlettertopic);
System.out.println(KafkaHeaders.RECEIVED_TOPIC);
libraryEventConsumerConfig.getTopic(topic);`
我在这里使用方法 getTopic 尝试自动创建 kafka 主题。在下面你可以看到 libraryEventConsumer class:
@Configuration
@EnableKafka
public class LibraryEventConsumerConfig {
@Bean
public void getTopic(String topic){
NewTopic deadlettertopic = TopicBuilder.name(String.format("%s_deadletter",topic))
.partitions(1)
.replicas(1)
.build();
}
}
不幸的是,该方法无效,我收到以下错误消息:
Parameter 0 of method getTopic in com.kafkalibrary.Config.LibraryEventConsumerConfig required a bean of type 'java.lang.String' that could not be found.
知道如何进行吗?
解决方案示例代码:
对于那些正在寻找相同目标的人,这是我的示例代码:感谢 Gary Russell 的启发。
private static void createTopic(String topicName, int numPartitions) throws Exception {
Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:5052,localhost:5053,localhost:5054");
AdminClient admin = AdminClient.create(config);
//checking if topic already exists
boolean alreadyExists = admin.listTopics().names().get().stream()
.anyMatch(existingTopicName -> existingTopicName.equals(topicName));
if (alreadyExists) {
System.out.printf("topic already exits: %s%n", topicName);
} else {
//creating new topic
System.out.printf("creating topic: %s%n", topicName);
NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) 1);
admin.createTopics(Collections.singleton(newTopic)).all().get();
}
添加再平衡侦听器,或扩展 AbstractConsumerSeekAware
(或仅实施 ConsumerSeekAware
)。
public class LibraryEventsConsumer extends AbstractConsumerSeekAware {
然后,在 onPartitionsAssigned()
中使用 AdminClient
检查 DLT 主题是否存在,如果不存在,则创建它。