package com.xiaopeng.automaticPilot.config;
|
|
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.beans.factory.annotation.Value;
|
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.core.*;
|
|
import java.util.HashMap;
|
import java.util.Map;
|
|
//TODO 这个配置类不生效的
|
/**
|
* Kafka 配置类
|
*/
|
@Configuration
|
@EnableKafka
|
public class KafkaConfig {
|
|
@Value("${spring.kafka.bootstrap-servers}")
|
private String bootstrapServers;
|
|
/**
|
* 生产者配置
|
*/
|
@Bean
|
public ProducerFactory<String, String> producerFactory() {
|
Map<String, Object> configProps = new HashMap<>();
|
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
configProps.put(ProducerConfig.RETRIES_CONFIG, 3);
|
configProps.put(ProducerConfig.ACKS_CONFIG, "all");
|
configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
|
configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
|
configProps.put(ProducerConfig.LINGER_MS_CONFIG, 1);
|
configProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
|
return new DefaultKafkaProducerFactory<>(configProps);
|
}
|
|
/**
|
* Kafka 模板
|
*/
|
@Bean
|
public KafkaTemplate<String, String> kafkaTemplate() {
|
return new KafkaTemplate<>(producerFactory());
|
}
|
|
/**
|
* 消费者配置
|
*/
|
@Bean
|
public ConsumerFactory<String, String> consumerFactory() {
|
Map<String, Object> props = new HashMap<>();
|
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
props.put(ConsumerConfig.GROUP_ID_CONFIG, "xp-automaticPilot-group");
|
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
|
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
|
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
|
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
|
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
|
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
|
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
|
return new DefaultKafkaConsumerFactory<>(props);
|
}
|
|
/**
|
* 监听器容器工厂
|
*/
|
@Bean
|
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
|
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
|
factory.setConsumerFactory(consumerFactory());
|
factory.setConcurrency(3);
|
factory.getContainerProperties().setPollTimeout(3000);
|
return factory;
|
}
|
}
|
