1、pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.5.4</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>kafkaDemo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>kafkaDemo</name>
<description>kafkaDemo</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2、application.properties
server.port=7780 server.servlet.context-path=/kafka
3、生产者和消费者
package com.example.kafkademo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
@RestController
@SpringBootApplication
public class KafkaDemoApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaDemoApplication.class, args);
}
@GetMapping("/send/{key}/{value}")
public String send(@PathVariable("key") String key, @PathVariable("value") String value) {
//配置信息
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "10.17.12.158:9092");
//设置数据key和value的序列化处理类
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
//创建生产者实例
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<String, String>("demo", key, value);
//发送记录
producer.send(record);
producer.close();
return "Ok";
}
@GetMapping("/receive/{topic}")
public void receive(@PathVariable("topic") String topic) {
System.out.println("topic="+topic);
//配置信息
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
//必须指定消费者组
props.put("group.id", "test");
//设置数据key和value的序列化处理类
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", StringDeserializer.class);
//创建消息者实例
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
//订阅topic1的消息
consumer.subscribe(Arrays.asList(topic));
//到服务器中读取记录
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(2000));
System.out.println("获取消息条数："+records.count());
for (ConsumerRecord<String, String> record : records) {
System.out.println("key:" + record.key() + "" + ",value:" + record.value());
}
}
}
}
4、编译Jar包
（1）在IDEA底部找到Terminal
（2）确认项目根目录，执行mvn clean package
（3）编译成功，看到BUILD SUCCESS
（4）找到编译好的jar包
[root@node1 ~]# ls jar kafkaDemo-0.0.1-SNAPSHOT.jar [root@node1 jar]# java -jar /root/jar/kafkaDemo-0.0.1-SNAPSHOT.jar . ____ _ __ _ _ /\\ / ___‘_ __ _ _(_)_ __ __ _ \ \ \ ( ( )\___ | ‘_ | ‘_| | ‘_ \/ _` | \ \ \ \\/ ___)| |_)| | | | | || (_| | ) ) ) ) ‘ |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v2.5.4) 2021-09-06 21:50:46.475 INFO 20561 --- [ main] c.e.kafkademo.KafkaDemoApplication : Starting KafkaDemoApplication v0.0.1-SNAPSHOT using Java 1.8.0_161 on node1 with PID 20561 (/root/jar/kafkaDemo-0.0.1-SNAPSHOT.jar started by root in /root/jar) 2021-09-06 21:50:46.493 INFO 20561 --- [ main] c.e.kafkademo.KafkaDemoApplication : No active profile set, falling back to default profiles: default 2021-09-06 21:50:48.467 INFO 20561 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat initialized with port(s): 7780 (http) 2021-09-06 21:50:48.494 INFO 20561 --- [ main] o.apache.catalina.core.StandardService : Starting service [Tomcat] 2021-09-06 21:50:48.495 INFO 20561 --- [ main] org.apache.catalina.core.StandardEngine : Starting Servlet engine: [Apache Tomcat/9.0.52] 2021-09-06 21:50:48.604 INFO 20561 --- [ main] o.a.c.c.C.[Tomcat].[localhost].[/kafka] : Initializing Spring embedded WebApplicationContext 2021-09-06 21:50:48.605 INFO 20561 --- [ main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 1981 ms 2021-09-06 21:50:49.283 INFO 20561 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 7780 (http) with context path ‘/kafka‘ 2021-09-06 21:50:49.301 INFO 20561 --- [ main] c.e.kafkademo.KafkaDemoApplication : Started KafkaDemoApplication in 3.594 seconds (JVM running for 4.398)