Spring Boot集成Kafka的示例代码

   4.0.0com.laravelshao.springbootspring-boot-integration-kafka0.0.1-SNAPSHOTjarspring-boot-integration-kafkaDemo project for Spring Boot org.springframework.bootspring-boot-starter-parent2.0.0.RELEASE UTF-8UTF-81.8  org.springframework.bootspring-boot-starter org.springframework.kafkaspring-kafka org.springframework.bootspring-boot-starter-json org.springframework.bootspring-boot-starter-testtest   org.springframework.bootspring-boot-maven-plugin

 spring: kafka: bootstrap-servers:X.X.X.X:9092 producer: value-serializer: org.springframework.kafka.support.serializer.JsonSerializer consumer: group-id: test auto-offset-reset: earliest value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring: json: trusted: packages: com.laravelshao.springboot.kafka 

 public class Message { private Integer id; private String msg; public Message() { } public Message(Integer id, String msg) { this.id = id; this.msg = msg; } public Integer getId() { return id; } public void setId(Integer id) { this.id = id; } public String getMsg() { return msg; } public void setMsg(String msg) { this.msg = msg; } @Override public String toString() { return "Message{" + "id=" + id + ", msg='" + msg + '\'' + '}'; } } 

 package com.laravelshao.springboot.kafka; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Component; /** * Created by shaoqinghua on 2018/3/23. */ @Component public class Producer { private static Logger log = LoggerFactory.getLogger(Producer.class); @Autowired private KafkaTemplate kafkaTemplate; public void send(String topic, Message message) { kafkaTemplate.send(topic, message); log.info("Producer->topic:{}, message:{}", topic, message); } } 

 package com.laravelshao.springboot.kafka; 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; /** * Created by shaoqinghua on 2018/3/23. */ @Component public class Consumer { private static Logger log = LoggerFactory.getLogger(Consumer.class); @KafkaListener(topics = "test_topic") public void receive(ConsumerRecord consumerRecord) { log.info("Consumer->topic:{}, value:{}", consumerRecord.topic(), consumerRecord.value()); } } 

 package com.laravelshao.springboot; import com.laravelshao.springboot.kafka.Message; import com.laravelshao.springboot.kafka.Producer; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ApplicationContext; @SpringBootApplication public class IntegrationKafkaApplication { public static void main(String[] args) throws InterruptedException { ApplicationContext context = SpringApplication.run(IntegrationKafkaApplication.class, args); Producer producer = context.getBean(Producer.class); for (int i = 1; i <10; i++) { producer.send("test_topic", new Message(i, "test topic message " + i)); Thread.sleep(2000); } } } 

 spring: kafka: consumer: properties: spring: json: trusted: packages: com.laravelshao.springboot.kafka