spring boot 与kafka集成的示例代码

<?xml version="1.0" encoding="UTF-8"?>

  4.0.0

  com.example
  demo
  0.0.1-SNAPSHOT
  jar

  demo
  Demo project for Spring Boot

  
    org.springframework.boot
    spring-boot-starter-parent
    1.5.8.RELEASE
     
  

  
    UTF-8
    UTF-8
    1.8
  

  
    
      org.springframework.boot
      spring-boot-starter-web
    

    
      org.springframework.kafka
      spring-kafka
    

    
      org.springframework.boot
      spring-boot-starter-test
      test
    
  

  
    
      
        org.springframework.boot
        spring-boot-maven-plugin
      
    
  

spring.kafka.bootstrap-servers=vm208:9092,vm:9092,vm50:9092
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.group-id=local_test
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=1

package com.example.demo.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class KafkaConsumer {

  private Logger logger = LoggerFactory.getLogger(this.getClass());

  @KafkaListener(topics = {"test"})
  public void listen(ConsumerRecord<?, ?> record) {
    System.out.printf("offset = %d,key =%s,value=%s\n", record.offset(), record.key(), record.value());
  }
}

package com.example.demo.producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class KafkaProducer {

  @Autowired
  private KafkaTemplate kafkaTemplate;
  String topic="test";
  public void sendMessage(String key,String data){
    kafkaTemplate.send(new ProducerRecord(topic,key,data));
  }
}

package com.example.demo.controller;
import com.example.demo.producer.KafkaProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class ProducerController {
  @Autowired
  private KafkaProducer kafkaProducer;
  @RequestMapping(value = "/kafka/send.do", method = RequestMethod.GET)
  public String sendMessage(@RequestParam(value = "key") String key, @RequestParam(value = "data") String data) {
    kafkaProducer.sendMessage(key, data);
    return "sucess";
  }
}