net域名做网站怎么样,黄骅市属于沧州市吗,装修公司加盟品牌,app商城需要手机网站吗1、简介 对于经常需要变更kafka主题的场景#xff0c;为了实现动态监听topic的功能#xff0c;可以使用以下方式。
2、使用步骤
2.1、添加依赖
dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactI…1、简介 对于经常需要变更kafka主题的场景为了实现动态监听topic的功能可以使用以下方式。
2、使用步骤
2.1、添加依赖
dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactIdversion2.8.1/version
/dependency
2.2、nacos中配置 # kafka 配置
spring:kafka:bootstrap-servers: ip地址:9092topics: topic1,tpic2producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerenable-idempotence: trueacks: alltransactional-id: kafka-groupconsumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: kafka-clickhouse-groupauto-offset-reset: latestenable-auto-commit: falseisolation-level: read_committedallow-auto-create-topics: truelistener:ack-mode: MANUAL_IMMEDIATEconcurrency: 3
2.3、配置类
package org.aecsi.kafkadatatock.config;import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.transaction.KafkaTransactionManager;import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;Configuration
RequiredArgsConstructor
EnableKafka
RefreshScope
public class KafkaConfig {private final KafkaProperties kafkaProperties;Beanpublic KafkaAdmin kafkaAdmin() {return new KafkaAdmin(kafkaProperties.buildAdminProperties());}Beanpublic AdminClient adminClient(KafkaAdmin kafkaAdmin) {return AdminClient.create(kafkaAdmin.getConfigurationProperties());}Beanpublic ProducerFactoryString, String producerFactory() {MapString, Object configProps new HashMap(kafkaProperties.buildProducerProperties());configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);configProps.put(ProducerConfig.ACKS_CONFIG, all);configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, kafka-clickhouse-producer);DefaultKafkaProducerFactoryString, String factory new DefaultKafkaProducerFactory(configProps);factory.setTransactionIdPrefix(kafka-clickhouse-producer-);return factory;}Beanpublic KafkaTemplateString, String kafkaTemplate(ProducerFactoryString, String producerFactory) {return new KafkaTemplate(producerFactory);}BeanRefreshScopepublic ConsumerFactoryString, String consumerFactory() {MapString, Object configProps new HashMap(kafkaProperties.buildConsumerProperties());configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);configProps.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, true);configProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, read_committed);return new DefaultKafkaConsumerFactory(configProps);}Beanpublic KafkaTransactionManagerString, String transactionManager(ProducerFactoryString, String producerFactory) {return new KafkaTransactionManager(producerFactory);}Beanpublic ConcurrentKafkaListenerContainerFactoryString, String kafkaListenerContainerFactory(ConsumerFactoryString, String consumerFactory,KafkaTransactionManagerString, String transactionManager) {ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory();factory.setConsumerFactory(consumerFactory);factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);factory.getContainerProperties().setTransactionManager(transactionManager);return factory;}BeanRefreshScopepublic ConcurrentKafkaListenerContainerFactoryString, String kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory();factory.setConsumerFactory(consumerFactory());factory.setAutoStartup(true);return factory;}Beanpublic ApplicationRunner kafkaListenerStarter(KafkaListenerEndpointRegistry registry) {return args - {// 启动所有 Kafka 监听器registry.start();};}
}接收消息类
KafkaListener(topics #{${spring.kafka.topics}.split(,)}, autoStartup false)Transactional(transactionManager transactionManager)public void processMessage(ConsumerRecordString, String record,Acknowledgment acknowledgment,Header(KafkaHeaders.RECEIVED_TOPIC) String topic,Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp) {try {log.info(kafka 接受 topic: {} 消息, topic);
// 处理消息acknowledgment.acknowledge();} catch (Exception e) {log.error(Error processing message for topic {}: {}, topic, e.getMessage());throw e;}} 主启动类添加一个注解
EnableConfigurationProperties(KafkaProperties.class)
3、总结 实现kafka动态获取topic还有其他方式博主目前只验证这一种其他方式待更新。
