1.首先在pom.xml中引入spring-kafka jar：

<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
</dependency>

2.创建kafkaListenerFactory和KafkaConsumerFactory，配置类如下：

@Configuration
@EnableKafka
public class KafkaConfig {

    private static final String KAFKA_SERVERS_CONFIG = "10.192.77.202:9092";
    private static final String LOCAL_GROUP_ID = "test";

    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS_CONFIG);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, LOCAL_GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS_CONFIG);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
}

3.之后便可通过@KafkaListener和kafkaTemplate对象监听和发送消息了：

@Controller
public class KafkaTestAction {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @RequestMapping("/admin/kafka/send")
    public String sendMsg(String msg){
        kafkaTemplate.send("TEST_TOPIC_NEW", msg);
        return "success";
    }

    @KafkaListener(topics = "TEST_TOPIC_NEW")
    public void listen(String data) {
        System.out.println("kafkaconfig =listen======="+data);
    }
}