java操作kafka实例，包括发送和接收任务

【实例简介】

基于springcloud的微服务工程，可以直接运行。

【实例截图】

【核心代码】

package kafkaserver;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
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.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

//\u901a\u8fc7\u6ce8\u89e3\u76d1\u542ctopic\u8fdb\u884c\u6d88\u8d39
@Configuration  
@EnableKafka
public class KafkaConsumer {

    final static String list = "192.168.201.185:9092,192.168.201.150:9092,192.168.201.96:9092";

    /**
     *  
     * Description\uff1a\u83b7\u53d6\u914d\u7f6e
     * Date\uff1a        2017\u5e747\u670811\u65e5
     * @author      shaqf
     */  
    private Map<String, Object> consumerConfigs() {
        Map<String, Object> props =  new HashMap();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, list);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);  
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");  
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");  
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);  
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);  
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //System.out.println("KafkaConsumer consumerConfigs " JsonUtils.objectToJson(props));
        return props;  
    }
    /** \u83b7\u53d6\u5de5\u5382 */
    private ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    /** \u83b7\u53d6\u5b9e\u4f8b */  
    @Bean  
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory1 = new ConcurrentKafkaListenerContainerFactory<>();
        factory1.setConsumerFactory(consumerFactory());
        factory1.setConcurrency(2);
        factory1.getContainerProperties().setPollTimeout(3000);  
        //System.out.println("KafkaConsumer kafkaListenerContainerFactory factory" JsonUtils.objectToJson(factory1));
        return factory1;
    }

    /**
     * topic\u7684\u6d88\u8d39\u8005\u7ec41\u76d1\u542c
     * @return
     */
    @Bean
    public Group1Listener listener1() {
        return new Group1Listener();
    }

    /**
     * topic\u7684\u6d88\u8d39\u8005\u7ec42\u76d1\u542c
     * @return
     */
    @Bean
    public Group2Listener listener2() {
        return new Group2Listener();
    }

}