| | |
| | | private int concurrency; |
| | | @Value("${kafka.groupId.second-data}") |
| | | private String secondDataGroupId; |
| | | @Value("${kafka.groupId.cruiser-data}") |
| | | private String cruiserDataGroupId; |
| | | |
| | | @Bean |
| | | public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() { |
| | |
| | | return factory; |
| | | } |
| | | |
| | | @Bean("cruiserDataListenerFactory") |
| | | public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> cruiserDataListenerFactory(){ |
| | | ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); |
| | | factory.setConsumerFactory(cruiserDataConsumerFactory());//设置消费者工厂 |
| | | factory.setConcurrency(concurrency);//设置线程数 |
| | | factory.getContainerProperties().setPollTimeout(1500);//设置拉取数据超时时间 |
| | | return factory; |
| | | } |
| | | |
| | | /** |
| | | * @Description: 秒级消费工厂 |
| | | * @Param: [] |
| | |
| | | public ConsumerFactory<String,String> secondDataConsumerFactory(){ |
| | | Map<String, Object> commonConfig = consumerConfigs(); |
| | | Map<String, Object> secondDataConfig = secondConsumerConfigs(); |
| | | secondDataConfig.putAll(commonConfig); |
| | | return new DefaultKafkaConsumerFactory<>(secondDataConfig); |
| | | } |
| | | |
| | | /* |
| | | * 走航车数据消费工厂 |
| | | * */ |
| | | public ConsumerFactory<String,String> cruiserDataConsumerFactory(){ |
| | | Map<String, Object> commonConfig = consumerConfigs(); |
| | | Map<String, Object> secondDataConfig = cruiserConsumerConfigs(); |
| | | secondDataConfig.putAll(commonConfig); |
| | | return new DefaultKafkaConsumerFactory<>(secondDataConfig); |
| | | } |
| | |
| | | return propsMap; |
| | | } |
| | | |
| | | /* |
| | | * 走航车消费组配置 |
| | | * */ |
| | | public Map<String,Object> cruiserConsumerConfigs(){ |
| | | Map<String, Object> propsMap = new HashMap<>(); |
| | | propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, cruiserDataGroupId); |
| | | return propsMap; |
| | | } |
| | | |
| | | /** |
| | | * @Description: 通用配置 |