浙江网站建设公司名单,wordpress开发的网站,网站加手机建设png图标,互联网推广策略使用Okta的身份管理平台轻松部署您的应用程序 使用Okta的API在几分钟之内即可对任何应用程序中的用户进行身份验证#xff0c;管理和保护。 今天尝试Okta。 当今的用户希望可以通过其计算机#xff0c;手机#xff0c;平板电脑或任何其他设备访问您的应用程序#xff01; … 使用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 languagejava \-d dependenciesweb,kafka \-d packageNamecom.okta.javakafka \-d namekafka-java \-d typemaven-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 ProducerFactoryString, String producerFactory() {return new DefaultKafkaProducerFactory(producerConfigurations());}Beanpublic MapString, Object producerConfigurations() {MapString, 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 KafkaTemplateString, 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 KafkaTemplateString, String template;public KafkaController(KafkaTemplateString, 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// localhost8080 / kafka / producemessage 这是我的消息 。 当您使用上述命令进行调用时您的应用程序将执行/kafka/produce端点该端点将消息发送到Kafka中的myTopic主题。 但是您如何知道该命令已成功向该主题发送了消息 现在您不会在应用程序内使用消息这意味着您不能确定 幸运的是有一种简单的方法可以立即创建消费者以进行测试。 在您的Kafka目录的bin文件夹中运行以下命令 ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic 访问http// localhost8080 / kafka / producemessage 这是我的消息再次在运行Kafka使用者的终端中看到以下消息 This is my message 做得好 您可以暂时停止此命令。 让我们添加一些Java代码来使用应用程序中的消息而不是从终端执行。 在Java应用中使用来自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 ConsumerFactoryString, String consumerFactory() {return new DefaultKafkaConsumerFactory(consumerConfigurations());}Beanpublic MapString, Object consumerConfigurations() {MapString, 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;}BeanConcurrentKafkaListenerContainerFactoryString, String kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactoryString, 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 ListString messages new ArrayList();KafkaListener(topics myTopic, groupId kafka-sandbox)public void listen(String message) {synchronized (messages) {messages.add(message);}}public ListString 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 KafkaTemplateString, String template;private MyTopicConsumer myTopicConsumer;public KafkaController(KafkaTemplateString, 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 ListString getMessages() {return myTopicConsumer.getMessages();}} 此类现在具有一个新的终结点以显示存储在使用者中的消息。 调用此端点时它将发送已从Kafka主题处理过的当前消息。 您的Java应用程序现在同时具有Kafka生产者和使用者因此让我们一起进行测试 重新启动您的应用程序然后转到http// localhost8080 / kafka / messages 。 目前没有信息被返回。 原因很简单您的使用者仅配置为接收新消息而您尚未发送新消息。 让我们通过访问网络浏览器并访问http// localhost8080 / kafka / producemessage 我的应用程序发送的消息来解决此问题 。 当Kafka收到该消息时它将立即让您的消费者知道它。 继续并在浏览器中转到http// localhost8080 / kafka / messages 。 现在您将看到您的消息已成功接收 做得好 您有一个能够从Kafka产生和使用消息的Java应用程序 但是在我们将其称为“一天”之前还有最后一步这是非常重要的一步。 保护您的Java Kafka应用程序 您的应用目前不是很安全。 尽管您已经准备好在分布式环境中处理许多消息但是那些可以找到指向您的端点的链接的人仍然可以使用这些消息。 这是一个关键漏洞因此请确保已正确解决此漏洞。 您将使用OAuth 2.0来确保只有经过身份验证的用户才能看到您的端点。 最好的部分 使用Okta验证用户身份只需5分钟即可在您的应用中添加此功能 创建一个Okta帐户 如果您还没有Okta帐户 请继续创建一个 。 完成注册后请执行以下步骤 登录到您的帐户 转到应用程序 添加应用程序 。 您将被重定向到以下页面 选择网站 然后单击下一步。 在表格中填写以下选项 姓名 Bootiful Kafka 点击完成 现在您有了Okta应用程序可以使用它来在Java Kafka应用程序中对用户进行身份验证。 使用用户身份验证保护Java应用安全 首先将Okta的库添加到您的项目中。 打开您的pom.xml并在dependencies标记内添加以下依赖项 dependencygroupIdcom.okta.spring/groupIdartifactIdokta-spring-boot-starter/artifactIdversion1.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_ISSUERhttps://{yourOktaDomain}/oauth2/default
export OKTA_OAUTH2_CLIENT_ID{yourClientID}
export OKTA_OAUTH2_CLIENT_SECRET{yourClientSecret} 您可以在Okta UI的应用程序页面中找到{yourClientID}和{yourClientSecret} 。 要访问它请按照以下步骤操作 在您的Okta菜单中转到“ 应用程序” 选择Bootiful Kafka应用程序 单击常规选项卡 您应该在“客户端凭据”区域内看到两个值。 值{yourOktaDomain}将在您的Okta仪表板中可见只需单击菜单上的仪表板。 您将在右上角看到组织URL。 而已 重新启动Spring Boot应用程序然后转到http// localhost8080 / kafka / messages 。 您的应用程序现在会将您重定向到登录页面 注意如果不提示您登录那是因为您已经登录。在隐身窗口中打开您的应用程序您将看到上面显示的登录屏幕。 输入您的用户名和密码。 如果登录尝试成功您将再次被重定向回您的应用程序。 恭喜你 您现在有了一个安全的Java应用程序该应用程序可以生成和使用来自Kafka的消息。 如果您想查看本教程的完整源代码请转到 GitHub上的oktadeveloper / okta-java-kafka-example 。 想更多地了解Java安全性和OAuth 2.0 以下是您可能感兴趣的一些链接 OAuth 2.0 Java指南5分钟保护您的应用程序安全 OAuth和OpenID Connect图解指南 使用Spring Boot和Kotlin构建应用程序 带有Spring Boot和Spring Cloud的Java微服务 使用Spring Cloud Gateway保护反应式微服务 有关此类文章的更多信息 请在Twitter上关注oktadev 。 我们还会定期将截屏视频发布到我们的YouTube频道 使用Okta的身份管理平台轻松部署您的应用程序 使用Okta的API在几分钟之内即可对任何应用程序中的用户进行身份验证管理和保护。 今天尝试Okta。 翻译自: https://www.javacodegeeks.com/2020/01/kafka-with-java-build-a-secure-scalable-messaging-app.html
