kafka服务端和springboot中使用

一、kafka-sever安装使用

下载kafka-server

https://kafka.apache.org/downloads.html

启动zookeeper

zookeeper-server-start.bat config\zookeeper.properties

启动kafka-server

kafka-server-start.bat config\server.properties

创建主题

kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

查看主题列表

kafka-topics.bat --list --bootstrap-server localhost:9092

创建生产者

kafka-console-producer.bat --broker-list localhost:9092 --topic test

在生产者控制台可以发消息

创建消费者

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

在消费者端可以收消息

查看主题详情

kafka-topics.bat --describe --bootstrap-server localhost:9092 --topic test

删除主题

kafka-delete-records.bat --bootstrap-server localhost:9092 --offset-json-file d:\delete_script.json

二、springboot中使用kafka

在Spring Boot中使用Kafka，你可以通过集成spring-kafka库来实现。这个库提供了对Apache Kafka的全面支持，包括生产者和消费者的功能。

添加依赖

首先，在你的pom.xml文件中添加以下依赖：

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

配置Kafka属性

在application.properties或application.yml文件中配置Kafka的相关属性。这里以application.properties为例：

# Kafka Producer Configuration
spring.kafka.producer.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer# Kafka Consumer Configuration
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

创建消息生产者

创建一个类用于发送消息到Kafka主题：

package space.goldchen.kafka.kafka;import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
/*** kafka生产者* @author 2021* @create 2025-04-16 16:27*/
@Service
@Slf4j
public class KafkaProducer {@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;/*** 发送消息* @param topic 主题* @param message 要发送的消息*/public void sendMessage(String topic, String message) {kafkaTemplate.send(topic, message);// 用log.info重写下边的日志log.info("Message sent to topic: {}, Message:{}" , topic, message);}
}

创建消息消费者

创建一个类用于从Kafka主题接收消息：

package space.goldchen.kafka.kafka;import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
/*** kafka消费者** @author 2021* @create 2025-04-16 16:28*/@Service
@Slf4j
public class KafkaConsumer {/*** my-topic 消费者消费消息* @param message*/@KafkaListener(topics = "my-topic", groupId = "my-consumer-group")public void consume(String message) {log.info("[my-topic]Received message: {}", message);}
}

测试Kafka集成

最后，编写一个简单的测试类来验证Kafka的生产和消费功能：

package space.goldchen.kafka.kafka;/*** kafka接口，调用生产者发送消息* @author 2021* @create 2025-04-16 16:28*/import jakarta.annotation.Resource;
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 {@Resourceprivate KafkaProducer kafkaProducer;/*** 使用KafkaProducer发送消息* @param message* @return*/@GetMapping("/send")public String sendMessage(@RequestParam String message) {kafkaProducer.sendMessage("my-topic", message);return "Message sent: " + message;}
}