哈尔滨建设网站公司哪家好,论坛网站设计,app在线设计,深汕特别合作区属于哪个区这里写自定义目录标题 springboot 配置Kafka 关闭自启动连接方法一#xff1a;使用 ConditionalOnProperty方法二#xff1a;手动管理Kafka监听器容器方法三#xff1a;使用 autoStartupfalse结语 springboot 配置Kafka 关闭自启动连接
在Spring Boot应用程序中#xff0c… 这里写自定义目录标题 springboot 配置Kafka 关闭自启动连接方法一使用 ConditionalOnProperty方法二手动管理Kafka监听器容器方法三使用 autoStartupfalse结语 springboot 配置Kafka 关闭自启动连接
在Spring Boot应用程序中默认情况下Kafka监听器容器会在应用程序启动时自动开始连接到Kafka broker。如果你希望禁用这种自动启动行为可以通过配置来实现。以下是几种常见的方法
方法一使用 ConditionalOnProperty
你可以使用条件注解来控制Kafka监听器容器的启动。通过设置一个属性来决定是否启用Kafka监听器。 步骤
定义配置属性 在你的application.yml或application.properties文件中添加一个自定义属性用于控制Kafka监听器的启用状态。 spring:kafka:enabled: false使用 ConditionalOnProperty 注解 在你的Kafka监听器类上使用ConditionalOnProperty注解根据配置属性来决定是否启用该监听器。 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Component;ComponentConditionalOnProperty(name spring.kafka.enabled, havingValue true)public class MyKafkaListener {KafkaListener(topics your-topic-name, groupId your-group-id)public void listen(String message) {System.out.println(Received Message: message);}}
方法二手动管理Kafka监听器容器
另一种方法是手动管理Kafka监听器容器的生命周期而不是依赖于Spring Boot的自动配置。 步骤
禁用自动配置 在你的主应用程序类或配置类上排除KafkaAutoConfiguration。 import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;SpringBootApplication(exclude KafkaAutoConfiguration.class)public class DeviceExchangeApplication {public static void main(String[] args) {long startTime System.currentTimeMillis();System.out.println(----------- 数据交换链[device-exchange]启动...);SpringApplication.run(DeviceExchangeApplication.class, args);System.out.println(----------- 数据交换链[device-exchange]启动成功耗时 (System.currentTimeMillis() - startTime) 毫秒);}}
手动创建和管理Kafka监听器容器 创建并管理Kafka监听器容器以便在需要的时候手动启动它们。 import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.core.ConsumerFactory;import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import org.springframework.kafka.listener.MessageListenerContainer;import java.util.HashMap;import java.util.Map;Configurationpublic class KafkaConfig {Autowiredprivate MyKafkaListener myKafkaListener;Beanpublic MapString, Object consumerConfigs() {MapString, Object props new HashMap();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.GROUP_ID_CONFIG, your-group-id);return props;}Beanpublic ConsumerFactoryString, String consumerFactory() {return new DefaultKafkaConsumerFactory(consumerConfigs());}Beanpublic ConcurrentKafkaListenerContainerFactoryString, String kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory();factory.setConsumerFactory(consumerFactory());return factory;}Beanpublic MessageListenerContainer kafkaListenerContainer() {ConcurrentMessageListenerContainerString, String container kafkaListenerContainerFactory().createContainer(your-topic-name);container.setupMessageListener(myKafkaListener::listen);return container;}}
手动启动Kafka监听器容器 在需要的时候手动启动Kafka监听器容器。 import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.CommandLineRunner;import org.springframework.kafka.listener.MessageListenerContainer;import org.springframework.stereotype.Component;Componentpublic class KafkaStarter implements CommandLineRunner {Autowiredprivate MessageListenerContainer kafkaListenerContainer;Overridepublic void run(String... args) throws Exception {// 手动启动Kafka监听器容器kafkaListenerContainer.start();}}
方法三使用 autoStartupfalse
你可以在Kafka监听器容器的配置中设置autoStartupfalse这样它就不会在应用程序启动时自动启动。 步骤
配置 autoStartupfalse 在你的Kafka监听器配置中设置autoStartupfalse。 import org.springframework.kafka.annotation.KafkaListener;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import org.springframework.kafka.listener.MessageListenerContainer;import org.springframework.stereotype.Component;Componentpublic class MyKafkaListener {KafkaListener(id myListener, topics your-topic-name, autoStartup false)public void listen(String message) {System.out.println(Received Message: message);}}
手动启动Kafka监听器容器 使用MessageListenerContainer接口的手动启动方法。 import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.listener.MessageListenerContainer;import org.springframework.stereotype.Component;Componentpublic class KafkaStarter {Autowiredprivate MessageListenerContainer myListenerContainer;public void startKafkaListener() {myListenerContainer.start();}}
总结 通过上述三种方法你可以有效地控制Kafka监听器容器的自动启动行为。选择适合你项目需求的方法来实现即可。通常情况下使用ConditionalOnProperty是最简单和灵活的方式。
结语
以上答案来自大模型第二种和第三种都比较麻烦最后采用了第一种方式在所有的消费类上加了ConditionalOnProperty(name spring.kafka.enabled, havingValue true)启动就很快了KafkaAdmin 和 KafkaConsumer就没有自动启动了。用kafkaTemplate发送消息还是会去连接Kafka服务器不影响正常使用。 注意必须是所有的消费类必须加不然就不会起作用。 主要场景一般线上部署环境才会去连接kafka本地开发的时候 不一定要去连所以想暂时关闭一下
