60-spring 消费kafka, 推送消费到kafka

spring 消费kafka

消费者配置：

@Configuration
@EnableKafka
@ConditionalOnResource(resources = "/special-run.txt")
public class ZdryKafkaConsumerConfig {

	@Value("${kafka.zdry.consumer.autoStartup}")
	private Boolean autoStartup;
	@Value("${kafka.zdry.consumer.servers}")
	private String servers;
	@Value("${kafka.zdry.consumer.topic}")
	private String topic;
	@Value("${kafka.zdry.consumer.group.id}")
	private String groupId;
	@Value("${kafka.zdry.consumer.enable.auto.commit}")
	private String enableAutoCommit;

	@Value("${kafka.zdry.consumer.auto.commit.interval.ms}")
	private String autoCommitIntervalMs;
	@Value("${kafka.zdry.consumer.session.timeout.ms}")
	private String sessionTimeoutMs;
	@Value("${kafka.zdry.consumer.auto.offset.reset}")
	private String autoOffsetReset;
	@Value("${kafka.zdry.consumer.max.poll.records}")
	private String maxPollRecords;
	@Value("${kafka.zdry.consumer.concurrency}")
	private Integer concurrency;


	/**
	 * 消费者批量工厂 人员轨迹
	 */
	@Bean("zdry_person_track")
	public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchFactory() {
		ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(consumerFactory());
		factory.setConcurrency(concurrency);
		factory.getContainerProperties().setPollTimeout(1500);
		factory.setBatchListener(true);
		factory.setAutoStartup(autoStartup);
		return factory;
	}

	/**
	 * 消费者工厂
	 */
	public ConsumerFactory<String, String> consumerFactory() {
		return new DefaultKafkaConsumerFactory<>(consumerConfigs());
	}

	/**
	 * 消费者配置
	 */
	public Map<String, Object> consumerConfigs() {
		Map<String, Object> propsMap = new HashMap<>();
		propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
		propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
		propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs);
		propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
		propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
		propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
		propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		return propsMap;
	}

}

消费监听：

@KafkaListener(topics = "${kafka.zdry.consumer.topic}", containerFactory = "zdry_person_track")
    public void consumeToJson(List<ConsumerRecord<String, String>> records){
        int count = records.size();
        log.info("count={}",count);
        JSONArray arr = new JSONArray();
        for (ConsumerRecord<String, String> record : records) {
            JSONObject obj = JSON.parseObject(record.value());
            arr.add(obj);
        }

    }

推送消费到kafka

生产者kafka配置：

@Configuration
@EnableKafka
@ConditionalOnResource(resources = "/special-run.txt")
public class ZdryKafkaProducerConfig {
    @Value("${kafka.zdry.consumer.servers}")
    private String servers;

    @Bean("zdryProducerTemplate")
    public KafkaTemplate<String, List<Object>> createTemplate(){
        Map<String, Object> pros = producerProps();
        ProducerFactory<String, List<Object>> pf = new DefaultKafkaProducerFactory<String, List<Object>>(pros);
        KafkaTemplate<String, List<Object>> template = new KafkaTemplate<>(pf);
        return template;
    }

    public Map<String, Object> producerProps() {

        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;

    }


}

产生消费并推送到kafka

    @Autowired
    private KafkaTemplate zdryProducerTemplate;

    @Value("${kafka.zdry.consumer.push.topic}")
    private String pushTopic;

方法：

try{
    zdryProducerTemplate.send(pushTopic,JSON.toJSONString(warn));
}catch(Exception e){
    e.printStackTrace();
}
zdryProducerTemplate.flush();

yml 配置

kafka:
  ## 重点人员抓拍轨迹
  zdry:
    consumer:
      autoStartup: true
      servers: x.x.x.x:6667
      topic: person_track
      group.id: aa_1
      enable.auto.commit: true
      auto.commit.interval.ms: 100
      session.timeout.ms: 10000
      auto.offset.reset: earliest
      max.poll.records: 100
      concurrency: 1
      push.topic: ai_warn