iot-transfer/src/main/java/com/persagy/iottransfer/config/KafkaCommonConfig.java

package com.persagy.iottransfer.config;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * kafka配置，实际上，在KafkaAutoConfiguration中已经有默认的根据配置文件信息创建配置，但是自动配置属性没有涵盖所有
 * 我们可以自定义创建相关bean，进行如下配置
 *
 * @author zhoujy
 * @date 2018年12月17日
 **/
@Configuration
@EnableConfigurationProperties({KafkaProperties.class})
@ConditionalOnProperty(prefix = "spring.kafka", name = "enable", havingValue = "true")
public class KafkaCommonConfig implements InitializingBean {

    private final KafkaProperties properties;
    @Value("${spring.kafka.consumer.topics}")
    private String topics;

    public KafkaCommonConfig(KafkaProperties properties) {
        this.properties = properties;
    }

    //构造消费者属性map，ConsumerConfig中的可配置属性比spring boot自动配置要多
    private Map<String, Object> consumerProperties() {
//        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        return properties.buildConsumerProperties();
    }

    /**
     * 不使用spring boot默认方式创建的DefaultKafkaConsumerFactory，重新定义创建方式
     *
     * @return
     */
    @Bean("consumerFactory")
    public DefaultKafkaConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory(consumerProperties());
    }


    @Bean("listenerContainerFactory")
    //个性化定义消费者
    public ConcurrentKafkaListenerContainerFactory listenerContainerFactory(DefaultKafkaConsumerFactory consumerFactory) {
        //指定使用DefaultKafkaConsumerFactory
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);

        //设置消费者ack模式为手动，看需求设置
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        //设置可批量拉取消息消费，拉取数量一次3，看需求设置
        factory.setConcurrency(1);
        factory.setBatchListener(true);
        return factory;
    }

   /* @Bean
     //代码创建方式topic
    public NewTopic batchTopic() {
        return new NewTopic("topic.quick.batch", 8, (short) 1);
    }*/

    //创建生产者配置map，ProducerConfig中的可配置属性比spring boot自动配置要多
    private Map<String, Object> producerProperties() {
        return properties.buildProducerProperties();
    }

    /**
     * 不使用spring boot的KafkaAutoConfiguration默认方式创建的DefaultKafkaProducerFactory，重新定义
     *
     * @return
     */
    @Bean("produceFactory")
    public DefaultKafkaProducerFactory produceFactory() {
        return new DefaultKafkaProducerFactory(producerProperties());
    }


    /**
     * 不使用spring boot的KafkaAutoConfiguration默认方式创建的KafkaTemplate，重新定义
     *
     * @param produceFactory
     * @return
     */
    @Bean
    public KafkaTemplate kafkaTemplate(DefaultKafkaProducerFactory produceFactory) {
        return new KafkaTemplate(produceFactory);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        String topicName = wireTopics();
        System.setProperty("topicName", topicName);
        System.out.println("### set system config topic:{}" + topicName);
    }

    private String wireTopics() {
        Set<String> topicSet = new HashSet<>();
        topicSet.add(topics);
        return StringUtils.join(topicSet, ",");
    }

}