SpringBoot——集成Kafka详解
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaProducerConfig {
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// 可以添加更多的配置属性
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
这段代码定义了一个配置类,其中包含了producerFactory
方法来创建一个ProducerFactory
实例,并且配置了Kafka生产者需要的基本属性,如服务地址和序列化器。kafkaTemplate
方法则使用这个ProducerFactory
创建了一个KafkaTemplate
实例,这个实例可以用来发送消息到Kafka。在Spring Boot应用中,你可以通过依赖注入来使用这个KafkaTemplate
实例。
评论已关闭