在 Spring Boot 2.7.x 中引入 Kafka-0.9 的实践

一、下载 Kafka-0.9

# 1. 下载
wget https://archive.apache.org/dist/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz# 2. 解压
tar -zxvf kafka_2.11-0.11.0.0.tgz# 3. 进入kafka目录
cd kafka_2.11-0.11.0.0

二、启动 Zookeeper 和 Kafka

# 1. 启动 zk 
bin/zookeeper-server-start.sh config/zookeeper.properties# 2. 启动 kafka
bin/kafka-server-start.sh config/server.properties# 3. 创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic user_topic# 4. 消费
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic user_topic --from-beginning

三、创建 Spring Boot 项目

https://blog.csdn.net/Agan__/article/details/136109762
https://start.spring.io/

四、引入 kafka 依赖

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>1.0.5.RELEASE</version>
</dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>0.9.0.0</version>
</dependency>

五、移除 Kafka 自动配置

@SpringBootApplication(exclude = {KafkaAutoConfiguration.class})

六、编写 Kafka 生产者

6.1 Kafka配置类

package com.chenjiacheng.samples.kafka.config;import com.chenjiacheng.samples.kafka.kafka.UserProducerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;import java.util.HashMap;
import java.util.Map;/*** Created by chenjiacheng on 2025/2/27 01:18** @author chenjiacheng* @since 1.0.0*/
@Slf4j
@Configuration
public class KafkaProducerConfig {@Autowiredprivate UserProducerListener userProducerListener;// 配置 User Topic 的 KafkaTemplate 和 ProducerListener@Bean(name = "userProducerFactory")public ProducerFactory<String, String> userProducerFactory() {Map<String, Object> configProps = new HashMap<>();configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return new DefaultKafkaProducerFactory<>(configProps);}@Bean(name = "userKafkaTemplate")public KafkaTemplate<String, String> userKafkaTemplate() {KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(userProducerFactory());kafkaTemplate.setProducerListener(userProducerListener);kafkaTemplate.setDefaultTopic("user");return kafkaTemplate;}
}

6.2 生产者监听类

package com.chenjiacheng.samples.kafka.kafka;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class UserProducerListener implements ProducerListener<String, String> {@Overridepublic void onSuccess(String topic, Integer partition, String key, String value, RecordMetadata recordMetadata) {log.info("User message sent successfully: {}", value);}@Overridepublic void onError(String topic, Integer partition, String key, String value, Exception exception) {log.error("Failed to send user message: {}", value, exception);}@Overridepublic boolean isInterestedInSuccess() {return true;}
}

七、编写Controller发送Kafka

package com.chenjiacheng.samples.kafka.controller;import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;@RestController
public class KafkaController {@Autowiredprivate KafkaTemplate<String, String> orderKafkaTemplate;@GetMapping("/send/user")public String sendUserMessage(@RequestParam("message") String message) {orderKafkaTemplate.sendDefault(message);return "User message sent: " + message;}
}

八、验证消费者

curl --location --globoff 'http://localhost:8081/send/user?message=hello'

Other. Spring Boot 引入 Kafka-0.11

其他操作同上, 仅修改依赖版本号. 也可直接使用 SpingBoot 依赖管理的 spring-kafka.

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>1.3.11.RELEASE</version>
</dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>0.11.0.0</version>
</dependency>