Spring Boot Kafka 生产者和消费者

2025 年 1 月 23 日 | 阅读 16 分钟

Spring Boot 是一个开源框架，旨在简化基于 Spring 的应用程序的开发。它提供了一种流线型的方式来创建独立的、生产级的 Spring 应用程序，并且配置最少。下面将详细介绍 Spring Boot。

什么是 Spring Boot？

Spring Boot 是更广泛的 Spring 生态系统的一部分，Spring 生态系统是用于企业 Java 开发的综合性框架。Spring Boot 的主要目标是轻松创建和部署独立的、生产级的 Spring 应用程序，而无需进行大量配置。

Spring Boot 的主要特性

1. 自动配置

Spring Boot 的自动配置功能会根据您包含的依赖项自动配置您的应用程序。例如，如果您包含数据库的依赖项，Spring Boot 将自动配置 DataSource。

2. Spring Initializr

这是一个基于 Web 的工具，允许开发人员快速生成具有必要依赖项的 Spring Boot 项目。它提供了一个简单的界面，您可以在其中选择项目类型、依赖项和其他设置。

3. 独立应用程序

Spring Boot 应用程序可以独立运行，这意味着它们可以使用简单的 main 方法作为独立的 Java 应用程序执行。这通常通过将 Tomcat 或 Jetty 等 Web 服务器直接嵌入到应用程序中来实现。

4. 生产就绪功能

Spring Boot 包含多项功能，可帮助在生产环境中监视和管理应用程序，例如指标、运行状况检查和外部化配置。

5. Spring Boot Starters

这些是预配置的依赖项，可简化向应用程序添加功能。例如，spring-boot-starter-web 包含了创建 Web 应用程序所需的所有依赖项。

6. Actuator

Spring Boot Actuator 提供了一套用于监视和管理 Spring Boot 应用程序的工具。它包括用于检查应用程序运行状况、查看指标等的端点。

Spring Boot 项目结构

典型的 Spring Boot 项目具有以下结构

src/main/java: 包含应用程序的 Java 源代码文件。
src/main/resources: 包含应用程序资源，例如属性文件。
src/test/java: 包含应用程序的测试 Java 文件。
pom.xml 或 build.gradle: 项目的构建配置文件，取决于您使用的是 Maven 还是 Gradle。

这是一个简单的 Spring Boot 应用程序主类的示例

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
public class MySpringBootApplication {
    public static void main(String[] args) {
        SpringApplication.run(MySpringBootApplication.class, args);
    }
}

Apache Kafka 概述

Apache Kafka 是一个由 LinkedIn 开发并捐赠给 Apache 软件基金会的开源流处理软件平台。它使用 Scala 和 Java 编写。Kafka 旨在处理实时数据馈送，并用于构建实时数据管道和流应用程序。下面将详细介绍 Kafka。

什么是 Apache Kafka？

Kafka 是一个分布式流处理平台，可以实时发布、订阅、存储和处理记录流。它具有高度的可扩展性和容错性，使其适合大规模数据处理应用程序。

核心概念

主题 (Topics): 主题是存储和发布记录的类别或馈送名称。Kafka 主题分布在多个节点上进行分区和复制，以实现可扩展性和容错性。
分区 (Partitions): 每个主题被划分为多个分区，这些分区是不断追加记录的有序、不可变序列。分区允许 Kafka 通过跨多个服务器分发负载来实现水平扩展。
代理 (Brokers): Kafka 集群由多个 Kafka 代理组成，每个代理负责管理一部分主题分区。代理负责接收、存储和提供数据。
生产者 (Producers): 生产者是向 Kafka 主题发布记录的应用程序。生产者可以写入任何主题和分区。
消费者 (Consumers): 消费者是订阅 Kafka 主题并处理记录的应用程序。消费者可以属于消费者组，以实现负载平衡和容错。
ZooKeeper: Kafka 使用 Apache ZooKeeper 进行分布式协调，例如管理代理元数据、分区的领导者选举和配置管理。

Kafka 与 Spring Boot 的集成

集成 Spring Boot 与 Kafka 的重要性

将 Kafka 与 Spring Boot 集成具有多种优势

易于使用: Spring Boot 抽象了设置和管理 Kafka 所涉及的许多复杂性，使开发人员更容易使用 Kafka。
可扩展性: Kafka 的分布式特性与 Spring Boot 的可扩展性相辅相成，使应用程序能够高效地处理大量数据。
可靠性: Kafka 强大的消息传递保证（至少一次、最多一次、精确一次）确保了可靠的数据处理，这对于关键应用程序至关重要。
灵活性: Spring Boot 的模块化以及 Kafka 与各种数据源和接收器集成的能力，使这种组合对于各种用例都具有高度的灵活性。

集成 Spring Boot 与 Kafka 的重要性

集成 Spring Boot 与 Kafka 结合了两者的优势，为开发可扩展、有弹性且实时处理数据的应用程序提供了一个强大的平台。以下是这种集成重要的几个原因：

简化的配置和开发
Spring Boot 的自动配置和启动器依赖项极大地减少了设置 Kafka 生产者或消费者所需的样板代码和配置。这使得开发人员可以更多地关注业务逻辑，而不是底层设置。
流线型的微服务架构
Spring Boot 被广泛用于开发微服务，而 Kafka 是处理服务间通信和数据流的绝佳选择。通过将 Kafka 与 Spring Boot 集成，开发人员可以构建通过 Kafka 主题高效可靠地通信的微服务，从而支持事件驱动的架构。
实时数据处理
Kafka 处理实时数据馈送的能力与 Spring Boot 的易开发性相结合，使得构建强大的实时数据处理应用程序成为可能。这在监视、分析和实时报告等场景中特别有用。
增强的可扩展性和容错性
Kafka 的分布式架构确保了高可扩展性和容错性，而 Spring Boot 应用程序可以轻松部署和扩展。这种集成确保应用程序可以处理大量数据并保持高可用性。
全面的监视和管理
Spring Boot Actuator 提供各种生产就绪的功能，例如运行状况检查和指标，可用于监视和管理 Kafka 生产者和消费者。这确保应用程序不仅易于开发，而且易于在生产环境中维护。

文档目标

本文档的主要目标是提供有关将 Spring Boot 与 Apache Kafka 集成以构建健壮且可扩展的数据处理应用程序的详细指南。这包括理论概念和实践示例，以帮助开发人员理解和实现使用 Spring Boot 的 Kafka 生产者和消费者。具体而言，本文档旨在：

解释基本概念: 提供对 Spring Boot 和 Apache Kafka 的深入理解，包括它们的主要特性、核心概念和典型用例。
分步设置: 提供设置开发环境的分步指南，包括安装 Kafka、配置 Spring Boot 以及创建 Kafka 生产者和消费者。
实践示例: 提供实际示例和代码片段，以演示如何在 Spring Boot 应用程序中实现 Kafka 生产者和消费者。这包括配置序列化器、处理错误和管理偏移量。
高级功能: 讨论 Kafka 的高级功能和配置，例如自定义序列化器、批量处理和消息确认。提供有关如何利用这些功能来处理更复杂用例的见解。
测试和最佳实践: 涵盖测试 Kafka 生产者和消费者的策略，包括单元测试、集成测试以及使用嵌入式 Kafka 进行测试。强调安全、资源管理和性能调优的最佳实践。
实际应用: 展示在各种场景（如微服务架构和实时数据处理系统）中集成 Spring Boot 与 Kafka 的实际示例和案例研究。
资源和进一步学习: 为希望深入了解 Spring Boot 和 Kafka 的读者建议其他资源，例如官方文档、书籍和在线课程。

使用 Spring Boot 构建 Kafka 生产者

将 Apache Kafka 与 Spring Boot 集成以创建 Kafka 生产者涉及多个步骤，包括配置必要的依赖项、设置应用程序属性、创建生产者以及最后将消息发送到 Kafka 主题。本节将提供每个步骤的详细演练，以帮助您在 Spring Boot 应用程序中设置 Kafka 生产者。

配置

首先，您需要将必要的 Kafka 依赖项添加到您的 Spring Boot 项目中。这可以使用 Maven 或 Gradle 来完成，具体取决于您选择的构建工具。

添加 Kafka 依赖项

对于 Maven 项目，请将以下依赖项添加到您的 pom.xml 文件中

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

对于 Gradle 项目，请将以下内容添加到您的 build.gradle 文件中

implementation 'org.springframework.kafka:spring-kafka:2.8.0'

配置生产者属性

接下来，您需要在 application.properties 或 application.yml 文件中配置 Kafka 生产者属性。这些属性定义了生产者将如何连接到 Kafka 集群以及如何处理消息生产。

application.properties

spring.kafka.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
spring.kafka.producer.acks=all
spring.kafka.producer.retries=10
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      retries: 10
      batch-size: 16384
      buffer-memory: 33554432

这些设置配置了以下内容：

bootstrap-servers: Kafka 代理的地址。
key-serializer: 键的序列化器类。
value-serializer: 值的序列化器类。
acks: 生产者在将请求视为完成之前要求领导者收到的确认数量。
retries: 发送失败的重试次数。
batch-size: 发送记录的批量大小。
buffer-memory: 生产者可用于缓冲等待发送到服务器的记录的总字节数。

创建生产者

在依赖项和配置到位后，下一步是在您的 Spring Boot 应用程序中创建 KafkaProducer bean。此 bean 将负责将消息发送到 Kafka 主题。

生产者配置类

创建一个配置类来定义 KafkaProducer bean

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
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.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");
        configProps.put(ProducerConfig.RETRIES_CONFIG, 10);
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        configProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

在此配置类中，producerFactory 创建了一个 ProducerFactory bean，KafkaTemplate bean 使用它将消息生产到 Kafka。KafkaTemplate 是一个高级 API，它包装了低级 Kafka 生产者 API，并简化了发送消息所需的代码。

将消息发送到 Kafka 主题

要将消息发送到 Kafka 主题，您将使用配置类中定义的 KafkaTemplate bean。创建一个服务类，它将使用 KafkaTemplate 发送消息。

KafkaProducerService

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    private static final String TOPIC = "my_topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
    }
}

预期输出

在此服务类中，sendMessage 方法使用 KafkaTemplate 将消息发送到指定的主题。主题名称设置为 my_topic。

示例：一个简单的 Spring Boot Kafka 生产者应用程序

为了提供一个完整的 Spring Boot Kafka 生产者应用程序示例，让我们将所有内容放在一起。此示例将包括主应用程序类、配置类、生产者服务以及触发消息发送的 REST 控制器。

项目结构

src
├── main
│   ├── java
│   │   └── com
│   │       └── example
│   │           └── kafka
│   │               |── KafkaProducerConfig.java
│   │               |── KafkaProducerService.java
│   │               |── KafkaProducerController.java
│   │               |── SpringBootKafkaApplication.java
│   |── resources
│       └── application.properties

SpringBootKafkaApplication.java

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

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

KafkaProducerConfig.java

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
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.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");
        configProps.put(ProducerConfig.RETRIES_CONFIG, 10);
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        configProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

KafkaProducerService.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    private static final String TOPIC = "my_topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
    }
}

KafkaProducerController.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaProducerController {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    @PostMapping("/publish")
    public String publishMessage(@RequestParam("message") String message) {
        kafkaProducerService.sendMessage(message);
        return "Message sent to Kafka topic successfully";
    }
}

application.properties

spring.kafka.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
spring.kafka.producer.acks=all
spring.kafka.producer.retries=10
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432

示例说明

SpringBootKafkaApplication: 这是 Spring Boot 应用程序的主类。它使用 SpringApplication.run 来启动应用程序。
KafkaProducerConfig: 此配置类定义了 KafkaProducer bean。它包括生产者属性，并创建一个 KafkaTemplate bean，将用于发送消息。
KafkaProducerService: 此服务类包含将消息发送到 Kafka 主题的业务逻辑。它使用 KafkaTemplate bean 将消息发送到指定的主题。
KafkaProducerController: 这个 REST 控制器提供了一个用于发布消息的端点。当向 /publish 端点发出带有 message 参数的 POST 请求时，它会调用 KafkaProducerService 的 sendMessage 方法将消息发送到 Kafka。
application.properties: 此属性文件包含 Kafka 生产者的配置设置，例如引导服务器和序列化器类。

运行应用程序

要运行应用程序，请确保您的 Kafka 代理正在 localhost:9092 上运行。然后，通过运行 SpringBootKafkaApplication 类来启动您的 Spring Boot 应用程序。您可以通过向带有 message 参数的 /publish 端点发出 POST 请求来发布消息。例如，您可以使用 curl 发送消息

这将将消息 "Hello Kafka" 发送到 my_topic Kafka 主题。

将消息发送到 Kafka 主题

要将消息发送到 Kafka 主题，您将使用配置类中定义的 KafkaTemplate bean。创建一个服务类，它将使用 KafkaTemplate 发送消息。

KafkaProducerService

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    private static final String TOPIC = "my_topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
    }
}

在此服务类中，sendMessage 方法使用 KafkaTemplate 将消息发送到指定的主题。主题名称设置为 my_topic。

使用 Spring Boot 构建 Kafka 消费者

在 Spring Boot 应用程序中设置 Kafka 消费者涉及几个关键步骤：配置消费者属性、创建消费者以及设置监听器以从 Kafka 主题消费消息。本节将提供每个步骤的详细演练，最终完成一个完整的 Spring Boot Kafka 消费者应用程序示例。

配置

首先，您需要将必要的 Kafka 依赖项添加到您的 Spring Boot 项目中。这可以使用 Maven 或 Gradle 来完成，具体取决于您选择的构建工具。

添加 Kafka 依赖项

对于 Maven 项目，请将以下依赖项添加到您的 pom.xml 文件中

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

对于 Gradle 项目，请将以下内容添加到您的 build.gradle 文件中

配置消费者属性

接下来，您需要在 application.properties 或 application.yml 文件中配置 Kafka 消费者属性。这些属性定义了消费者将如何连接到 Kafka 集群以及如何处理消息消费。

application.properties

spring.kafka.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

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

这些设置配置了以下内容：

bootstrap-servers: Kafka 代理的地址。
group-id: 此消费者所属的消费者组 ID。
auto-offset-reset: 确定当 Kafka 中没有初始偏移量或当前偏移量不存在时（例如，由于数据删除）该怎么做。 earliest 值表示消费者将从主题的开头开始读取。
key-deserializer: 键的反序列化器类。
value-deserializer: 值的反序列化器类。

创建消费者

在依赖项和配置到位后，下一步是在您的 Spring Boot 应用程序中创建 KafkaListener。此监听器将负责从 Kafka 主题消费消息。

消费者配置类

创建一个配置类来定义消费者工厂和 Kafka 监听器容器工厂

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

预期输出

在此配置类中，consumerFactory 方法创建了一个 ConsumerFactory bean，kafkaListenerContainerFactory bean 使用它来创建监听器容器以处理 Kafka 消息。@EnableKafka 注解用于启用 Kafka 监听器注解。

创建 KafkaListener

要从 Kafka 主题消费消息，您需要创建一个类，其中包含一个用 @KafkaListener 注解的方法。每当收到消息时，都会调用此方法。

KafkaConsumerService

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    @KafkaListener(topics = "my_topic", groupId = "my-group")
    public void listen(String message) {
        System.out.println("Received Message in group my-group: " + message);
    }
}

在此服务类中，listen 方法用 @KafkaListener 注解，指定要监听的主题和消费者组。当消息发布到 my_topic 主题时，将调用此方法，并将消息打印到控制台。

增强 Kafka 消费者应用程序

为了使 Kafka 消费者应用程序更健壮且适合生产环境，您可以实施多项增强和最佳实践。这些包括错误处理、消息确认、日志记录和性能调优。让我们逐一详细介绍这些增强功能。

错误处理

正确的错误处理可确保您的应用程序能够优雅地处理消息消费期间发生的异常。您可以通过在 KafkaListenerContainerFactory 中配置错误处理程序来实现错误处理。

Enhanced KafkaConsumerConfig.java

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, StringDeserializer.class.getName());
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setErrorHandler(errorHandler());
        return factory;
    }

    @Bean
    public ErrorHandler errorHandler() {
        return new SeekToCurrentErrorHandler();
    }
}

在此配置中

ErrorHandlingDeserializer 用于包装反序列化器并处理反序列化错误。
配置了 SeekToCurrentErrorHandler 以通过查找当前偏移量来处理错误。

消息确认

Kafka 支持不同的确认模式，以确保消息得到可靠处理。默认情况下，消息会被自动确认。您可以更改此行为，以便在处理后手动确认消息。

手动确认示例

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    @KafkaListener(topics = "my_topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            System.out.println("Received Message: " + record.value());
            // Process the message
            acknowledgment.acknowledge(); // Manually acknowledge
        } catch (Exception e) {
            // Handle exception
            e.printStackTrace();
        }
    }
}

在此示例中，listen 方法包含一个 Acknowledgment 参数。通过 acknowledge() 手动确认消息，这允许您控制何时认为消息已成功处理。

日志记录

日志记录对于监视和调试您的应用程序至关重要。使用 Logback 或 Log4j 等日志框架来捕获消息消费的详细日志。

Logback 配置 (logback.xml)

<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss} - %msg%n</pattern>
        </encoder>
    </appender>
    
    <logger name="org.apache.kafka" level="INFO"/>
    <logger name="org.springframework.kafka" level="INFO"/>
    
    <root level="DEBUG">
        <appender-ref ref="STDOUT"/>
    </root>
</configuration>

KafkaConsumerService.java 中的日志记录

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    @KafkaListener(topics = "my_topic", groupId = "my-group")
    public void listen(String message) {
        logger.info("Received Message: {}", message);
        // Process the message
    }
}

这种设置确保所有相关信息（包括消息消费和任何错误）都会被记录下来以备将来参考。

ConcurrentKafkaListenerContainerFactory 配置

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(3); // Increase concurrency
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
}

application.properties

spring.kafka.consumer.fetch-min-size=50000
spring.kafka.consumer.max-poll-records=100

增加并发性并调整获取和轮询设置可以显著提高 Kafka 消费者的吞吐量和效率。

完整的增强示例

项目结构

src
├── main
│   ├── java
│   │   └── com
│   │       └── example
│   │           └── kafka
│   │               ├── KafkaConsumerConfig.java
│   │               ├── KafkaConsumerService.java
│   │               └── SpringBootKafkaApplication.java
│   └── resources
│       ├── application.properties
│       └── logback.xml

SpringBootKafkaApplication.java

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

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

KafkaConsumerConfig.java

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, StringDeserializer.class.getName());
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.setErrorHandler(errorHandler());
        return factory;
    }

    @Bean
    public ErrorHandler errorHandler() {
        return new SeekToCurrentErrorHandler();
    }
}

KafkaConsumerService.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    @KafkaListener(topics = "my_topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            logger.info("Received Message: {}", record.value());
            // Process the message
            acknowledgment.acknowledge();
        } catch (Exception e) {
            logger.error("Error processing message", e);
        }
    }
}

application.properties

spring.kafka.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
spring.kafka.consumer.fetch-min-size=50000
spring.kafka.consumer.max-poll-records=100

logback.xml

<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss} - %msg%n</pattern>
        </encoder>
    </appender>
    
    <logger name="org.apache.kafka" level="INFO"/>
    <logger name="org.springframework.kafka" level="INFO"/>
    
    <root level="DEBUG">
        <appender-ref ref="STDOUT"/>
    </root>
</configuration>

Kafka 生产者和消费者反序列化

理解序列化和反序列化

序列化是将对象转换为字节流的过程，从而可以轻松地存储或传输对象。反序列化是相反的过程，其中字节流被转换回原始对象的副本。这些过程在 Apache Kafka 等分布式系统中至关重要，在这些系统中，数据必须跨不同的节点和服务进行传输。

Kafka 中的序列化

在 Kafka 中，序列化对于生产者至关重要，因为 Kafka 主题只能存储字节数组。这意味着您想要发送到 Kafka 主题的任何数据都必须首先序列化为字节数组。

Kafka 中的常见序列化器

StringSerializer: 将字符串转换为字节数组。
ByteArraySerializer: 用于原始字节数据。
AvroSerializer: 使用 Avro 格式（包括模式信息）序列化数据。
JSONSerializer: 将对象转换为 JSON 格式的字节数组。
Custom Serializers: 允许您为自定义对象定义自己的序列化逻辑。

序列化器的选择取决于数据格式和应用程序的要求。例如，如果您处理 JSON 数据，JSON 序列化器将是合适的。

Kafka 中的反序列化

Kafka 消费者使用反序列化将字节数组转换回原始对象。这使得消费者能够以有意义的方式处理数据。

Kafka 中的常见反序列化器

StringDeserializer: 将字节数组转换回字符串。
ByteArrayDeserializer: 用于原始字节数据。
AvroDeserializer: 反序列化 Avro 格式的数据。
JSONDeserializer: 将 JSON 字节数组转换回对象。
Custom Deserializers: 允许您为自定义对象定义自己的反序列化逻辑。

与序列化一样，反序列化器的选择取决于您 Kafka 主题中使用的数据格式以及应用程序的需求。

Kafka 中序列化和反序列化如何工作

当 Kafka 生产者发送消息时，它使用序列化器将消息对象转换为字节数组。然后，该字节数组通过网络发送到 Kafka 代理，Kafka 代理将其存储在指定的主题中。

当 Kafka 消费者从主题读取消息时，它使用反序列化器将字节数组转换回应用程序可以使用的对象。

序列化过程

生产者应用程序创建一个对象（例如，字符串、JSON 对象）。
对象被传递给生产者的序列化器。
序列化器将对象转换为字节数组。
字节数组被发送到 Kafka 代理并存储在主题中。

反序列化过程

消费者从 Kafka 主题读取字节数组。
字节数组被传递给消费者的反序列化器。
反序列化器将字节数组转换回对象。
对象由消费者应用程序处理。

下一主题事件流架构的用例

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Spring Boot Kafka 生产者和消费者

什么是 Spring Boot？

Spring Boot 的主要特性

1. 自动配置

2. Spring Initializr

3. 独立应用程序

4. 生产就绪功能

5. Spring Boot Starters

6. Actuator

Spring Boot 项目结构

Apache Kafka 概述

什么是 Apache Kafka？

Kafka 与 Spring Boot 的集成

集成 Spring Boot 与 Kafka 的重要性

集成 Spring Boot 与 Kafka 的重要性

文档目标

使用 Spring Boot 构建 Kafka 生产者

配置

配置生产者属性

创建生产者

生产者配置类

将消息发送到 Kafka 主题

示例：一个简单的 Spring Boot Kafka 生产者应用程序

SpringBootKafkaApplication.java

KafkaProducerConfig.java

KafkaProducerService.java

KafkaProducerController.java

application.properties

运行应用程序

将消息发送到 Kafka 主题

使用 Spring Boot 构建 Kafka 消费者

配置

配置消费者属性

创建消费者

消费者配置类

创建 KafkaListener

KafkaConsumerService

增强 Kafka 消费者应用程序

错误处理

Enhanced KafkaConsumerConfig.java

消息确认

手动确认示例

日志记录

KafkaConsumerService.java 中的日志记录

ConcurrentKafkaListenerContainerFactory 配置

完整的增强示例

项目结构

SpringBootKafkaApplication.java

KafkaConsumerConfig.java

KafkaConsumerService.java

application.properties