语言 Chinese, Simplified

SEO Title

How to Work with Apache Kafka in Your Spring Boot Application

在体系结构规划期间选择正确的消息传递系统始终是一个挑战，但这是需要确定的最重要的考虑因素之一。作为一名开发人员，我每天都要编写需要服务大量用户并实时处理大量数据的应用程序。

通常，我将Java与Spring框架(Spring Boot、Spring数据、Spring云、Spring缓存等)一起使用。Spring Boot是一个框架，它允许我比以前更快更轻松地完成开发过程。它已在我的组织中发挥了关键作用。随着用户数量的快速增长，我们意识到我们显然需要每秒处理1,000,000个事件。

当我们发现Apache Kafka®时，我们发现它满足了我们的需求，可以快速处理数百万条消息。这就是为什么我们决定尝试一下。从那一刻起，卡夫卡就成了我口袋里的重要工具。你会问，我为什么选择它?

Apache Kafka是:

可伸缩的
容错
一个很棒的发布-订阅消息传递系统
与大多数消息传递系统相比，具有更高的吞吐量
高度耐用
高度可靠
高的性能

这就是为什么我决定在我的项目中使用它。根据我的经验，我在这里提供了一个循序渐进的指南，介绍如何在Spring启动应用程序中包含Apache Kafka，以便您也可以开始利用它的优点。

先决条件

本文要求您拥有Confluent平台
手动安装使用ZIP和TAR档案
下载
- 解压缩它
- 按照逐步说明，您将在本地环境中启动和运行Kafka

我建议在您的开发中使用Confluent CLI来启动和运行Apache Kafka和流平台的其他组件。

你会从这本指南中得到什么

阅读完本指南后，您将拥有一个Spring Boot应用程序，其中包含一个Kafka生成器，用于向您的Kafka主题发布消息，以及一个Kafka使用者，用于读取这些消息。

好了，让我们开始吧!

表的内容

步骤1:生成项目
步骤2:发布/读取来自Kafka主题的消息
步骤3:通过应用程序配置Kafka。yml配置文件
步骤4:创建一个生产者
第五步:创造一个消费者
步骤6:创建一个REST控制器

步骤1:生成项目

首先，让我们使用Spring Initializr来生成我们的项目。我们的项目将有Spring MVC/web支持和Apache Kafka支持。

Generate a Spring Boot project with Spring Initializr

一旦你解压缩了这个项目，你将会有一个非常简单的结构。我将在本文的最后向您展示项目的外观，以便您能够轻松地遵循相同的结构。我将使用Intellij IDEA，但是你可以使用任何Java IDE。

步骤2:发布/读取来自Kafka主题的消息

现在，你可以看到它是什么样的。让我们继续讨论来自Kafka主题的发布/阅读消息。

首先创建一个简单的Java类，我们将使用它作为示例:package com.demo.models;

public class User {

private String name;
private int age;

public User(String name, int age) {
this.name = name;
this.age = age;
}
}

Spring Boot with Kafka – Intellij IDEA

步骤3:通过应用程序配置Kafka.yml配置文件

接下来，我们需要创建配置文件。我们需要以某种方式配置我们的Kafka生产者和消费者，使他们能够发布和从主题读取消息。我们可以使用任意一个应用程序，而不是创建一个Java类，并用@Configuration注释标记它。属性文件或application.yml。Spring Boot允许我们避免过去编写的所有样板代码，并为我们提供了更智能的配置应用程序的方法，如下所示:

server: port: 9000
spring:
kafka:
consumer:
bootstrap-servers: localhost:9092
group-id: group_id
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer:
bootstrap-servers: localhost:9092
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer

如果您想了解更多关于Spring引导自动配置的信息，可以阅读这篇简短而有用的文章。有关可用配置属性的完整列表，请参阅官方文档。

步骤4:创建一个生产者

创建生产者将把我们的消息写入主题。

@Service
public class Producer {

private static final Logger logger = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "users";

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String message) {
logger.info(String.format("#### -> Producing message -> %s", message));
this.kafkaTemplate.send(TOPIC, message);
}
}

我们只是自动连接KafkaTemplate，并将使用此实例发布消息到主题——这就是生产者!

第五步:创造一个消费者

Consumer是负责根据您自己的业务逻辑的需要读取消息并对其进行处理的服务。要设置它，请输入以下内容:

@Service
public class Consumer {

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

@KafkaListener(topics = "users", groupId = "group_id")
public void consume(String message) throws IOException {
logger.info(String.format("#### -> Consumed message -> %s", message));
}
}

在这里，我们告诉我们的方法void consumption (String message)订阅用户的主题，并将每个消息发送到应用程序日志。在实际的应用程序中，可以按照业务需要的方式处理消息。

步骤6:创建一个REST控制器

如果我们已经有了一个消费者，那么我们就已经拥有了消费Kafka消息所需的一切。

为了完整地显示我们创建的所有内容是如何工作的，我们需要创建一个具有单个端点的控制器。消息将被发布到这个端点，然后由我们的生产者进行处理。

然后，我们的使用者将以登录到控制台的方式捕获和处理它。

@RestController
@RequestMapping(value = "/kafka")
public class KafkaController {

private final Producer producer;

@Autowired
KafkaController(Producer producer) {
this.producer = producer;
}

@PostMapping(value = "/publish")
public void sendMessageToKafkaTopic(@RequestParam("message") String message) {
this.producer.sendMessage(message);
}
}

让我们用cURL把信息发送给Kafka: