1、本文架构
本文目标是使用微服务Spring Cloud集成Kafka实现异步通信。其中Kafka Server部署在Ubuntu虚拟机上，微服务部署在Windows 11系统上，Kafka Producer微服务和Kafka Consumer微服务分别注册到Eureka注册中心。Kafka Producer和Kafka Consumer之间通过Kafka Server实现异步通信。
出于便于测试的目的，我通过浏览器触发Kafka Producer发送消息，观察Kafka Consumer的后台是否打印出接收到的消息内容。
Ubuntu 上部署Kafka Server，详见博文：Ubuntu下Kafka安装及使用-CSDN博客
Eureka注册中心的搭建过程和完整代码，详见博文：微服务1：搭建微服务注册中心（命令行简易版，不使用IDE）-CSDN博客
Kafka Producer微服务的完整代码，详见博文：使用微服务Spring Cloud集成Kafka实现异步通信-CSDN博客
本文的重点是实现下图中的深蓝色部分：Kafka Consumer微服务。
2、创建Spring boot项目（Kafka Consumer微服务项目）：
mvn archetype:generate -DgroupId=com.test -DartifactId=microservice-kafka-consumer -DarchetypeArtifactId=maven-archetype-quickstart
项目代码的完整目录如下图所示：
编辑pom.xml，添加依赖包：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.test</groupId>
<artifactId>microservice-kafka-consumer</artifactId>
<packaging>jar</packaging>
<version>1.0-SNAPSHOT</version>
<name>microservice-kafka-consumer</name>
<url>http://maven.apache.org</url>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.0.RELEASE</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>Hoxton.SR4</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
编辑application.yml，配置kafka消费者：
consumer:
#消费的主题
topic: test-topic
#消费者组id
group-id: test-group
#是否自动提交偏移量
enable-auto-commit: true
#提交偏移量的间隔-毫秒
auto-commit-ms: 1000
#客户端消费的会话超时时间-毫秒
session-timeout-ms: 10000
#实现DeSerializer接口的反序列化类键
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
#实现DeSerializer接口的反序列化类值
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
server:
port: 8030
spring:
application:
name: microservice-kafka-consumer
kafka:
bootstrap-servers: 192.168.23.131:9092
consumer:
group-id: test-group
enable-auto-commit: true
auto-commit-ms: 1000
session-timeout-ms: 10000
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
eureka:
client:
serviceUrl:
defaultZone: http://localhost:8080/eureka/
instance:
prefer-ip-address: true
App.java的完整代码如下：
package com.test;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.kafka.annotation.KafkaListener;
@SpringBootApplication
@EnableDiscoveryClient
public class App
{
@KafkaListener(topics = "mydemo1")
public void listen(String msg) throws Exception {
System.out.println( "-----> Recv a msg: " + msg );
}
public static void main( String[] args ){
System.out.println( "Hello World!" );
SpringApplication.run(App.class, args);
}
}
3、测试
在浏览器输入，触发Kafka Producer向Kafka Server发送消息：
http://localhost:8020/kafka/sendMsg?msg=测试消息testmsg在Kafka Consumer的后台打印出收到的消息：
