| | |
| | | import java.util.HashMap; |
| | | import java.util.Map; |
| | | |
| | | /*@Configuration |
| | | @EnableKafka*/ |
| | | @Configuration |
| | | @EnableKafka |
| | | public class KafkaConsumerConfig { |
| | | @Value("${kafka.consumer.servers}") |
| | | private String servers; |
| | |
| | | private boolean enableAutoCommit; |
| | | @Value("${kafka.consumer.session.timeout}") |
| | | private String sessionTimeout; |
| | | @Value("${kafka.consumer.auto.commit.interval}") |
| | | private String autoCommitInterval; |
| | | @Value("${kafka.consumer.groupMenu.id}") |
| | | private String groupId; |
| | | @Value("${kafka.consumer.auto.offset.reset}") |
| | | private String autoOffsetReset; |
| | | @Value("${kafka.consumer.concurrency}") |
| | |
| | | @Bean |
| | | public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() { |
| | | ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); |
| | | factory.setConsumerFactory(consumerFactory()); |
| | | factory.setConcurrency(concurrency); |
| | | factory.getContainerProperties().setPollTimeout(1500); |
| | | factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); |
| | | |
| | | /*factory.setBatchListener(true);//@KafkaListener 批量消费 每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG |
| | | factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置提交偏移量的方式*/ |
| | | factory.setConsumerFactory(consumerFactory());//设置消费者工厂 |
| | | factory.setConcurrency(concurrency);//设置线程数 |
| | | factory.getContainerProperties().setPollTimeout(1500);//设置拉取数据超时时间 |
| | | return factory; |
| | | } |
| | | |
| | |
| | | Map<String, Object> propsMap = new HashMap<>(); |
| | | propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); |
| | | propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit); |
| | | propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval); |
| | | propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout); |
| | | propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); |
| | | propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); |
| | | propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); |
| | | propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); |
| | | return propsMap; |
| | | } |