java程序连接kafka_Java的Kafka：构建安全，可扩展的消息传递应用程序

java程序连接kafka

使用Okta的身份管理平台轻松部署您的应用程序 使用Okta的API在几分钟之内即可对任何应用程序中的用户进行身份验证，管理和保护。今天尝试Okta。

当今的用户希望可以通过其计算机，手机，平板电脑或任何其他设备访问您的应用程序！这种向软件即服务（SaaS）规范的过渡要求开发人员有效地与强大的工具集成，这些工具可以扩展为每秒处理数千（甚至数百万）个请求。 Apache Kafka是处理那些高吞吐量环境的最有效工具之一。

在本教程中，您将学习Apache Kafka的基本概念，并构建一个功能齐全的Java应用程序，该应用程序能够产生和使用来自Kafka的消息。

先决条件： Java 8+，互联网连接和免费的Okta开发人员帐户。

Apache Kafka的简要概述

Apache Kafka是一个分布式流媒体平台，它利用发布/订阅消息模式与应用程序进行交互，并且旨在创建持久消息。

让我们更详细地分解这些概念。

分布式流媒体平台

当您要运行Kafka时，您需要启动其代理：与其他任何服务器一样，在计算机上运行的Kafka的简单实例。代理负责将消息发送，接收和存储到磁盘中。

一个经纪人不足以确保Kafka可以处理高吞吐量的消息。该目标是通过许多经纪人同时合作，相互沟通和协调来实现的。

Kafka集群将一个或多个经纪人组合在一起。您的应用程序连接到一个群集，该群集为您管理所有分布式详细信息，而不是连接到单个节点。

具有持久消息的发布/订阅消息系统

发布/订阅是分布式系统中的常见模式。下图说明了Kafka中此模式的基本结构：

该图像包括到目前为止尚未提及的两个组件：生产者和消费者。

生产者是将消息发送到群集的应用程序。在此示例中，生产者1、2和3正在发送消息。然后，集群选择应由哪个代理存储它们，并将其发送给选定的代理。

另一方面，您有消费者。使用者是连接到集群并接收生产者发布的消息的应用程序。任何有兴趣使用生产者发送的消息的应用程序都必须连接到Kafka消费者。

由于Kafka会长时间存储消息（默认值为7天），因此即使发送消息时不在场，您也可以让许多使用者收到相同的消息！

卡夫卡主题

将消息发送到Kafka代理时，需要通过指定主题来指定将消息发送到的位置。主题是消费者可以订阅的消息类别。该机制确保使用者仅接收与其相关的消息，而不是接收发布到集群的每条消息。

现在您已经了解了Kafka的基本架构，让我们下载并安装它。

安装并运行Kafka

要下载Kafka，请访问Kafka网站。将此压缩文件的内容提取到您喜欢的文件夹中。

在Kafka目录中，转到bin文件夹。在这里，您会发现许多bash脚本，这些脚本对于运行Kafka应用程序很有用。如果使用的是Windows，则windows文件夹中也有相同的脚本。本教程使用Linux命令，但是如果您正在运行Microsoft OS，则只需使用等效的Windows版本。

启动Zookeeper管理您的Kafka群集

Apache Kafka始终作为分布式应用程序运行。这意味着您的集群必须在同步配置或选举负责人的过程中处理一些分布式挑战。

Kafka使用Zookeeper跟踪这些细节。不过，不必担心下载它。 Kafka已经与Zookeeper一起提供，可以让您快速启动并运行。

让我们启动一个Zookeeper实例！在您的Kafka目录中的bin文件夹中，运行以下命令：

./zookeeper-server-start.sh ../config/zookeeper.properties

默认情况下，此命令在端口2181上启动Zookeeper服务器。 Zookeeper负责协调集群内的Kafka经纪人。您将在本教程的Kafka项目中使用默认配置，但始终可以根据需要更改这些值。

运行卡夫卡经纪人

下一步是运行代理本身。在另一个终端上，从bin文件夹运行以下命令：

./kafka-server-start.sh ../config/server.properties

您可能已经猜到了，此命令在默认端口9092上以默认配置运行Kafka服务器。

创建一个Kafka主题

现在您已运行代理和Zookeeper，现在可以指定一个主题以开始从生产者发送消息。您将在bin文件夹中运行命令，就像前面的步骤一样：

./kafka-topics.sh --create --topic myTopic -zookeeper \localhost:2181 --replication-factor 1 --partitions 1

此命令创建一个名为myTopic ，该myTopic指向您使用第一个命令启动的Zookeeper实例。您还必须指定两个不同的参数： replication-factor和partitions 。现在不用担心它们–它们用于控制与Kafka中的分布式系统相关的特定方面。在运行简单设置时，您可以为两个参数指定“ 1”。

现在您已完成所有事情并可以运行，您可以开始将Kafka与Java应用程序集成！

创建一个Java + Kafka应用程序

让我们从项目结构开始，使用Spring Initializer创建应用程序。

转到https://start.spring.io并填写以下信息：

项目：Maven项目
语言：Java
群组： com.okta.javakafka
工件： kafka-java
依存关系：
- Spring网

您也可以使用命令行生成项目。将以下命令粘贴到您的终端中，它将下载与上面定义的配置相同的项目：

curl https://start.spring.io/starter.zip -d language=java \-d dependencies=web,kafka \-d packageName=com.okta.javakafka \-d name=kafka-java \-d type=maven-project \-o kafka-java.zip

本教程使用Maven，但是您可以根据需要轻松地使用Gradle进行学习。

而已！现在，您的Java项目结构已创建，您可以开始开发应用程序了。

在Java应用程序中将消息推送到Kafka主题

创建可以推送消息的生产者的第一步是在Java应用程序中配置生产者。让我们创建一个配置类来做到这一点。

创建一个src/main/java/com/okta/javakafka/configuration文件夹，并在其中创建一个ProducerConfiguration类：

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 java.util.HashMap;
import java.util.Map;@Configuration
public class ProducerConfiguration {private static final String KAFKA_BROKER = "localhost:9092";@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigurations());}@Beanpublic Map<String, Object> producerConfigurations() {Map<String, Object> configurations = new HashMap<>();configurations.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);configurations.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configurations.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return configurations;}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}}

此类创建一个ProducerFactory ，该ProducerFactory知道如何根据您提供的配置来创建生产者。您还指定了连接到本地Kafka代理，并使用String序列化密钥和值。

您还声明了一个KafkaTemplate bean在生产者上执行高级操作。换句话说，模板能够执行诸如将消息发送到主题之类的操作，并有效地向您隐藏了后台信息。

下一步是创建端点，以将消息发送给生产者。在src/main/java/com/okta/javakafka/controller软件包中，创建以下类：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import java.util.List;@RestController
public class KafkaController {private KafkaTemplate<String, String> template;public KafkaController(KafkaTemplate<String, String> template) {this.template = template;}@GetMapping("/kafka/produce")public void produce(@RequestParam String message) {template.send("myTopic", message);}}

注意：由于您要发送要处理的数据，所以produce()方法实际上应该是POST。出于演示目的，将其保留为GET更容易，因此您可以在浏览器中进行练习。

如您所见，此端点非常简单。它喷射KafkaTemplate前面配置和发送一个消息给myTopic当GET请求到由/kafka/produce 。

让我们测试一切是否按预期工作。在JavaKafkaApplication类中运行main方法。要从命令行运行，请执行以下命令：

./mvnw spring-boot:run

您的服务器应该在端口8080上运行，并且您已经可以对它发出API请求！

转到网络浏览器，然后访问http：// localhost：8080 / kafka / produce？message =这是我的消息。

当您使用上述命令进行调用时，您的应用程序将执行/kafka/produce端点，该端点将消息发送到Kafka中的myTopic主题。

但是您如何知道该命令已成功向该主题发送消息？现在，您不会在应用程序内使用消息，这意味着您不能确定！

幸运的是，有一种简单的方法可以创建一个立即进行测试的消费者。在您的Kafka目录的bin文件夹中，运行以下命令：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic

访问http：// localhost：8080 / kafka / produce？message =这是我的消息，再次在运行Kafka使用者的终端中看到以下消息：

This is my message

很好！您可以暂时停止此命令。

让我们添加一些Java代码来使用应用程序中的消息，而不是从终端执行。

在Java App中使用来自Kafka主题的消息

与生产者一样，您需要添加配置以使消费者能够找到Kafka经纪人。

在src/main/java/com/okta/javakafka/configuration创建以下类：

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.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import java.util.HashMap;
import java.util.Map;@Configuration
public class ConsumerConfiguration {private static final String KAFKA_BROKER = "localhost:9092";private static final String GROUP_ID = "kafka-sandbox";@Beanpublic ConsumerFactory<String, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigurations());}@Beanpublic Map<String, Object> consumerConfigurations() {Map<String, Object> configurations = new HashMap<>();configurations.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);configurations.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);configurations.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configurations.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);return configurations;}@BeanConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;}}

上面的代码创建了一个工厂，该工厂知道如何连接到本地代理。它还将您的使用者配置为针对键和值对String反序列化，以匹配生产者配置。

组ID是必需的，Kafka使用组ID来允许并行数据消耗。 ConcurrentKafkaListenerContainerFactory bean使您的应用程序可以在多个线程中使用消息。

现在，您的Java应用已配置为在Kafka经纪人中查找使用者，让我们开始收听发送给该主题的消息。

创建一个src/main/java/com/okta/javakafka/consumer目录，并在其中创建以下类：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;import java.util.ArrayList;
import java.util.List;@Component
public class MyTopicConsumer {private final List<String> messages = new ArrayList<>();@KafkaListener(topics = "myTopic", groupId = "kafka-sandbox")public void listen(String message) {synchronized (messages) {messages.add(message);}}public List<String> getMessages() {return messages;}}

此类负责侦听myTopic主题内的更改。它通过使用KafkaListener注释来实现。每当生产者向主题发送新消息时，您的应用程序都会在此类内接收到一条消息。它将一条消息添加到接收到的消息列表中，通过getMessages()方法使其可用于其他类。

接下来，让我们创建一个显示已消费消息列表的端点。返回KafkaController以添加MyTopicConsumer作为依赖项和getMessages()方法。

import com.okta.javakafka.consumer.MyTopicConsumer;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import java.util.List;@RestController
public class KafkaController {private KafkaTemplate<String, String> template;private MyTopicConsumer myTopicConsumer;public KafkaController(KafkaTemplate<String, String> template, MyTopicConsumer myTopicConsumer) {this.template = template;this.myTopicConsumer = myTopicConsumer;}@GetMapping("/kafka/produce")public void produce(@RequestParam String message) {template.send("myTopic", message);}@GetMapping("/kafka/messages")public List<String> getMessages() {return myTopicConsumer.getMessages();}}

此类现在具有一个新的终结点，以显示存储在使用者中的消息。调用此端点时，它将发送已从Kafka主题处理过的当前消息。

您的Java应用程序现在同时具有Kafka生产者和使用者，因此让我们一起进行测试！重新启动您的应用程序，然后转到http：// localhost：8080 / kafka / messages 。

目前，没有信息被返回。原因很简单：您的使用者仅配置为接收新消息，而您尚未发送新消息。让我们通过访问网络浏览器并访问http：// localhost：8080 / kafka / produce？message =我的应用程序发送的消息来解决此问题！。

当Kafka收到该消息时，它将立即让您的消费者知道它。继续并在浏览器中转到http：// localhost：8080 / kafka / messages 。现在，您将看到您的消息已成功收到！

很好！您有一个Java应用程序，能够产生和使用来自Kafka的消息！但是，在我们将其称为“一天”之前，还有最后一步，这是非常重要的一步。

保护您的Java Kafka应用程序

您的应用目前不是很安全。尽管您已经准备好在分布式环境中处理许多消息，但是那些可以找到指向您的端点的链接的人仍然可以使用这些消息。这是一个关键漏洞，因此请确保已正确解决此漏洞。

您将使用OAuth 2.0来确保只有经过身份验证的用户才能看到您的端点。最好的部分？使用Okta验证用户身份，只需5分钟即可在您的应用中添加此功能！

创建一个Okta帐户

如果您还没有Okta帐户，请继续创建一个。完成注册后，请执行以下步骤：

登录到您的帐户
转到应用程序 > 添加应用程序 。您将被重定向到以下页面：

选择网站，然后单击下一步。
在表格中填写以下选项：
- 姓名： Bootiful Kafka
点击完成

有了Okta应用程序之后，就可以使用它来在Java + Kafka应用程序中对用户进行身份验证。

使用用户身份验证保护Java应用安全

首先，将Okta的库添加到您的项目中。打开您的pom.xml并在<dependencies>标记内添加以下依赖项：

<dependency><groupId>com.okta.spring</groupId><artifactId>okta-spring-boot-starter</artifactId><version>1.3.0</version>
</dependency>

该库将与您刚创建的Okta应用集成。它还会将Spring Security添加到您当前的应用程序中。在src/main/resources/application.properties使用以下变量配置它：

okta.oauth2.issuer: https://{yourOktaDomain}/oauth2/default
okta.oauth2.client-id: {yourClientID}
okta.oauth2.client-secret: {yourClientSecret}

重要说明 ：此文件只能在本地使用。不要将客户的机密泄露给Git或任何其他版本控制系统。

为避免意外暴露这些凭据，您还可以将Okta应用程序的值指定为环境变量。使用以下环境变量在应用程序的根目录中创建okta.env文件。然后在启动您的应用程序之前运行source okta.env
export OKTA_OAUTH2_ISSUER=https://{yourOktaDomain}/oauth2/default
export OKTA_OAUTH2_CLIENT_ID={yourClientID}
export OKTA_OAUTH2_CLIENT_SECRET={yourClientSecret}

您可以在Okta UI的应用程序页面中找到{yourClientID}和{yourClientSecret} 。要访问它，请按照以下步骤操作：