Spring boot 集成Kafka

<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>
<groupId>com.java</groupId>
<artifactId>kafka</artifactId>
<version>1.0.0</version> <parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.5.RELEASE</version>
</parent> <dependencies> <!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency> <!-- 热部署 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>springloaded</artifactId>
<version>1.2.8.RELEASE</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>provided</scope>
</dependency> </dependencies> <build>
<finalName>${project.artifactId}</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin> <plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

package com.java;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;/**
* 主启动类
*
* @author Logan
* @version 1.0.0
* @createDate 2019-05-07
*
*/
@SpringBootApplication
public class KafkaStarter { public static void main(String[] args) {
SpringApplication.run(KafkaStarter.class, args);
}}

package com.java.listener;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/**
*
* 任务处理器，监听kafka队列中的消息，消费并处理
*
* @author Logan
* @version 1.0.0
* @createDate 2019-05-07
*
*/
@Component
public class MessageHandler { @KafkaListener(topics = { "test-topic" })
public void handle(String message) {
System.out.println("[ 处理器开始处理消息 ]" + System.currentTimeMillis()); try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
} System.out.println(message); System.out.println("[ 处理器处理消息完成 ]" + System.currentTimeMillis());
} @KafkaListener(topics = { "test-topic" })
public void handle(ConsumerRecord<String, String> record) {
System.out.println("[ 处理器开始处理消息 ]" + System.currentTimeMillis()); try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
} System.out.println(record); System.out.println("[ 处理器处理消息完成 ]" + System.currentTimeMillis());
}}

package com.java.controller;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;/**
* 发送消息类
*
* @author Logan
* @version 1.0.0
* @createDate 2019-05-07
*
*/
@RestController
public class SendMessageController { @Autowired
private KafkaTemplate<String, String> kafkaTemplate; private String topic = "test-topic"; @GetMapping("/send")
public String send(String params) {
System.out.println("[ 收到请求 ]"); kafkaTemplate.send(topic, params); System.out.println("[ 返回响应 ]");
return "您的任务已提交";
}}

#生产者配置，参见org.springframework.boot.autoconfigure.kafka.KafkaProperties.Producer
spring.kafka.producer.bootstrapServers=s1:9092,s2:9092,s3:9092#消费者配置，参见org.springframework.boot.autoconfigure.kafka.KafkaProperties.Consumer
spring.kafka.consumer.bootstrapServers=s1:9092,s2:9092,s3:9092
spring.kafka.consumer.groupId=kafka-test
spring.kafka.consumer.autoOffsetReset=latest
spring.kafka.consumer.enableAutoCommit=true