SpringBoot整合SpringCloudStream3.1+版本的Kafka死信队列

**Spring Boot 整合 Spring Cloud Stream3.1 + Kafka 死信队列**

在微服务架构中，消息队列是实现系统间通信的关键组件之一。Kafka 是一个流行的分布式消息队列系统，而 Spring Cloud Stream 则提供了对多种消息队列系统（包括 Kafka）的抽象接口和工具。Spring Boot 可以轻松整合 Spring Cloud Stream 和 Kafka 来构建高性能、可扩展的微服务系统。

在本文中，我们将重点介绍如何使用 Spring Boot 整合 Spring Cloud Stream3.1 + 版本的Kafka 死信队列。

### **依赖配置**

首先，我们需要在 `pom.xml` 文件中添加必要的依赖：

xml<dependencies>
 <!-- Spring Boot -->
 <dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-web</artifactId>
 </dependency>

 <!-- Spring Cloud Stream -->
 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-stream-binder-kafka</artifactId>
 </dependency>

 <!-- Kafka -->
 <dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka-clients</artifactId>
 </dependency>

 <!-- Spring Boot Starter Test -->
 <dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-test</artifactId>
 <scope>test</scope>
 </dependency>
</dependencies>

propertiesspring:
 cloud:
 stream:
 binder:
 kafka:
 bootstrap-servers: localhost:9092 default-key-schema: string default-value-schema: stringserver:
 port:8080

java@SpringBootApplication@EnableBinding(Sink.class)
public class KafkaProducerApplication {

 @Autowired private ProducerTemplate producerTemplate;

 public static void main(String[] args) {
 SpringApplication.run(KafkaProducerApplication.class, args);
 }

 @Bean public Function filterFunction() {
 return message -> {
 // Filter logic here return message;
 };
 }
}

java@SpringBootApplication@EnableBinding(Source.class)
public class KafkaConsumerApplication {

 @Autowired private Source source;

 public static void main(String[] args) {
 SpringApplication.run(KafkaConsumerApplication.class, args);
 }

 @Bean public Function filterFunction() {
 return message -> {
 // Filter logic here return message;
 };
 }
}

propertiesspring:
 cloud:
 stream:
 binder:
 kafka:
 bootstrap-servers: localhost:9092 default-key-schema: string default-value-schema: stringserver:
 port:8080kafka:
 topic:
 dead-letter-queue:
 name: dlq-topic

java@SpringBootApplication@EnableBinding(Sink.class)
public class KafkaProducerApplication {

 @Autowired private ProducerTemplate producerTemplate;

 public static void main(String[] args) {
 SpringApplication.run(KafkaProducerApplication.class, args);
 }

 @Bean public Function filterFunction() {
 return message -> {
 // Filter logic here if (message.equals("dead-letter")) {
 throw new RuntimeException("Dead letter exception");
 }
 return message;
 };
 }
}

java@SpringBootApplication@EnableBinding(Source.class)
public class KafkaConsumerApplication {

 @Autowired private Source source;

 public static void main(String[] args) {
 SpringApplication.run(KafkaConsumerApplication.class, args);
 }

 @Bean public Function filterFunction() {
 return message -> {
 // Filter logic here if (message.equals("dead-letter")) {
 throw new RuntimeException("Dead letter exception");
 }
 return message;
 };
 }
}