【Spring底层原理高级进阶】Spring Kafka：实时数据流处理，让业务风起云涌！️

 🎉🎉欢迎光临🎉🎉

🏅我是苏泽，一位对技术充满热情的探索者和分享者。🚀🚀

🌟特别推荐给大家我的最新专栏《Spring 狂野之旅：从入门到入魔》 🚀

本专栏带你从Spring入门到入魔！

这是苏泽的个人主页可以看到我其他的内容哦👇👇

努力的苏泽icon-default.png?t=N7T8http://suzee.blog.csdn.net/

 

故事引言

当我们谈论 Spring Kafka 时，可以把它想象成一位非常出色的邮递员，但不是运送普通的信件，而是处理大量的有趣和有用的数据。这位邮递员擅长与 Kafka 进行互动，并且以一种高级抽象和易用的方式处理数据。

这位邮递员的任务是将数据从一个地方传送到另一个地方，就像我们寄送包裹一样。他知道如何与 Kafka 进行通信，了解如何与输入和输出主题建立联系。

当有人将数据放入输入主题时，这位邮递员会立即接收到通知，并迅速将数据取出。然后，他会对这些数据进行各种有趣的转换和处理操作，就像是一个巧手的魔术师一样。他可以将数据转换成不同的格式、进行聚合、过滤、连接和分流等操作。

一旦数据处理完毕，这位邮递员会将数据装入一个特殊的包裹，并标上目的地的地址，这个目的地就是输出主题。然后，他会快速地把包裹发送出去，确保数据能够按时到达。

Spring Kafka 就像是这位邮递员的工具箱，提供了许多有用的工具和功能，使他的工作更加轻松。它提供了简单且声明性的 API，让我们可以用一种直观的方式定义数据的处理逻辑和流处理拓扑。

那么正文开始

目录

故事引言

简介和背景：

实时数据流处理对业务至关重要的原因：

Spring Kafka 基础知识：

深入了解 Apache Kafka 的核心概念和组件：

消息发布和消费：

消费者组管理：

消费者组的概念和作用：

实现有效的消费者组管理：以下是一些实现有效消费者组管理的关键考虑因素：

具体业务实践： 

流处理与处理拓扑

Kafka Streams 的概念和特性：

使用 Spring Kafka 构建和部署流处理拓扑：

实践：

首先，在 pom.xml 文件中添加以下 Maven 依赖：

然后，创建一个 Spring Kafka 流处理应用程序：

简介和背景：

Spring Kafka 是 Spring Framework 提供的一个集成 Apache Kafka 的库，用于构建基于 Kafka 的实时数据流处理应用程序。Apache Kafka 是一个高性能、分布式的流数据平台，广泛用于构建可扩展的、实时的数据处理管道。

实时数据流处理对业务至关重要的原因：

实时数据流处理对于现代业务来说非常重要。随着互联网的快速发展和数字化转型的加速，企业面临着大量的数据产生和处理的挑战。实时数据流处理能够帮助企业实时地捕获、处理和分析数据，从而使企业能够做出及时的决策、提供个性化的服务和优化业务流程。实时数据流处理还可以帮助企业发现潜在的机会和风险，并迅速采取行动。

Spring Kafka 基础知识：

深入了解 Apache Kafka 的核心概念和组件：

在开始学习 Spring Kafka 之前，了解 Apache Kafka 的核心概念和组件是非常重要的。一些核心概念包括：

  • 主题（Topic）：消息的类别或者主题。
  • 分区（Partition）：主题被分成多个分区，每个分区都是有序的，并且可以在多个机器上进行复制。
  • 生产者（Producer）：负责将消息发布到 Kafka 主题。
  • 消费者（Consumer）：从 Kafka 主题订阅并消费消息。
  • 消费者组（Consumer Group）：一组消费者共同消费一个或多个主题，每个主题的分区被分配给一个消费者组中的一个消费者。
  • 偏移量（Offset）：消费者可以跟踪已消费的消息的位置，通过偏移量来表示。

介绍 Spring Kafka 的基本用法和集成方式：

Spring Kafka 提供了简单而强大的 API，用于在 Spring 应用程序中使用 Kafka。它提供了以下核心功能：

  • 消息生产：使用 Spring Kafka 的 KafkaTemplate 类可以方便地将消息发布到 Kafka 主题。
  • 消息消费：通过使用 Spring Kafka 提供的 @KafkaListener 注解，可以轻松地创建消息消费者，并处理来自 Kafka 主题的消息。
  • 错误处理：Spring Kafka 提供了灵活的错误处理机制，可以处理消息发布和消费过程中的各种错误情况。
  • 事务支持：Spring Kafka 支持与 Spring 的事务管理机制集成，从而实现消息发布和消费的事务性操作。

消息发布和消费：

在 Spring Kafka 中发布消息到 Kafka 主题，你可以使用 KafkaTemplate 类的 send() 方法。通过指定要发送的主题和消息内容，可以将消息发送到 Kafka。

要消费 Kafka 主题中的消息，你可以使用 @KafkaListener 注解来创建一个消息监听器。通过指定要监听的主题和消息处理方法，可以在接收到消息时触发相应的逻辑。

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;public void publishMessage(String topic, String message) {kafkaTemplate.send(topic, message);
}

要消费 Kafka 主题中的消息，你可以使用 @KafkaListener 注解来创建一个消息监听器。通过指定要监听的主题和消息处理方法，可以在接收到消息时触发相应的逻辑。

@KafkaListener(topics = "myTopic")
public void consumeMessage(String message) {// 处理接收到的消息System.out.println("Received message: " + message);
}

理解消息的序列化和反序列化：

在 Kafka 中，消息的序列化和反序列化是非常重要的概念。当消息被发送到 Kafka 时，它们需要被序列化为字节流。同样地，在消息被消费时，它们需要被反序列化为原始的数据格式。

Spring Kafka 提供了默认的序列化和反序列化机制，可以根据消息的类型自动进行转换。对于常见的数据类型，如字符串、JSON、字节数组等，Spring Kafka 已经提供了相应的序列化和反序列化实现。此外，你也可以自定义序列化和反序列化器来处理特定的消息格式。

例如，你可以使用 StringSerializer 和 StringDeserializer 来序列化和反序列化字符串消息：

@Configuration
public class KafkaConfig {@Beanpublic ProducerFactory<String, String> producerFactory() {Map<String, Object> config = new HashMap<>();config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return new DefaultKafkaProducerFactory<>(config);}@Beanpublic ConsumerFactory<String, String> consumerFactory() {Map<String, Object> config = new HashMap<>();config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);return new DefaultKafkaConsumerFactory<>(config);}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;}
}

消费者组管理：

消费者组的概念和作用：

消费者组是一组具有相同消费者组ID的消费者，它们共同消费一个或多个 Kafka 主题的消息。消费者组的作用是实现消息的并行处理和负载均衡。通过将主题的分区分配给消费者组中的不同消费者，可以实现消息的并行处理，提高处理吞吐量和降低延迟。消费者组还提供了容错性，当某个消费者出现故障时，其他消费者可以接管其分区并继续处理消息。

实现有效的消费者组管理：
以下是一些实现有效消费者组管理的关键考虑因素：

  1. 消费者组ID的选择：为每个消费者组选择一个唯一的ID，确保不同的消费者组之间互不干扰。

  2. 分区分配策略：选择适当的分区分配策略，确保分配给消费者的分区负载均衡，并避免某些消费者负载过重或空闲。

  3. 动态扩缩容：根据负载情况和处理需求，动态地增加或减少消费者的数量，以实现弹性的消费者组管理。

  4. 监控和健康检查：监控消费者组的运行状态，及时发现并处理故障消费者，确保消费者组的稳定运行。

具体业务实践： 

假设有一个在线电商平台，用户可以在平台上购买商品。平台需要处理用户的订单，并将订单信息发送到一个 Kafka 主题中。订单处理包括验证订单、生成发货单、更新库存等操作。

在这个场景中，可以使用消费者组来实现订单处理的并行处理和负载均衡。具体步骤如下：

  1. 创建一个名为"order"的 Kafka 主题，用于接收用户的订单信息。

  2. 创建一个消费者组，比如名为"order-processing-group"的消费者组。

  3. 启动多个消费者实例，加入到"order-processing-group"消费者组中。每个消费者实例都会订阅"order"主题，并独立地消费订单消息。

  4. Kafka 会根据消费者组的配置，将"order"主题的分区均匀地分配给消费者组中的消费者实例。每个消费者实例将独立地处理分配给它的分区上的订单消息。

  5. 当有新的订单消息到达"order"主题时，Kafka 会将消息分配给消费者组中的一个消费者实例。消费者实例会处理订单消息，执行验证、生成发货单、更新库存等操作。

具体实现：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class OrderConsumer {private static final String TOPIC = "order";private static final String GROUP_ID = "order-processing-group";private static final String BOOTSTRAP_SERVERS = "localhost:9092";public static void main(String[] args) {// 创建消费者配置Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 创建 Kafka 消费者KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 订阅主题consumer.subscribe(Collections.singletonList(TOPIC));// 消费消息while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {String orderMessage = record.value();// 执行订单处理操作，例如验证订单、生成发货单、更新库存等processOrder(orderMessage);}}}private static void processOrder(String orderMessage) {// 实现订单处理逻辑System.out.println("Processing order: " + orderMessage);// TODO: 执行订单处理的具体业务逻辑}
}

​​​​​​​流处理与处理拓扑

  1. Kafka Streams 的概念和特性：

    • Kafka Streams 是一个用于构建实时流处理应用程序的客户端库。
    • 它允许开发人员以简单且声明性的方式处理 Kafka 主题中的数据流。
    • Kafka Streams 提供了丰富的功能，包括数据转换、数据聚合、窗口操作、连接和分流等。 
       // 创建拓扑建造器StreamsBuilder builder = new StreamsBuilder();// 创建输入流KStream<String, String> inputStream = builder.stream("input-topic");// 进行数据转换和处理操作KStream<String, String> outputStream = inputStream.mapValues(value -> value.toUpperCase()).filter((key, value) -> value.startsWith("A"));// 将处理结果输出到输出主题outputStream.to("output-topic");// 创建 Kafka Streams 实例KafkaStreams streams = new KafkaStreams(builder.build(), props);

    • 它具有高度可扩展性和容错性，可以通过水平扩展来处理大规模的数据流。
    • Kafka Streams 库紧密集成了 Kafka 的生态系统，可以无缝整合其他 Kafka 组件和工具。

  2. 使用 Spring Kafka 构建和部署流处理拓扑：

    • Spring Kafka 是 Spring Framework 提供的用于与 Kafka 交互的模块。
    • 它提供了高级抽象和易用的 API，简化了 Kafka 流处理应用程序的开发和集成。
    • 使用 Spring Kafka，可以通过配置和注解来定义流处理拓扑，包括输入和输出主题、数据转换和处理逻辑等。
    • Spring Kafka 还提供了与 Spring Boot 的集成，简化了应用程序的配置和部署流程。

实践：

首先，在 pom.xml 文件中添加以下 Maven 依赖：

<dependencies><!-- Spring Kafka 相关依赖 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.8.1</version></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka-test</artifactId><version>2.8.1</version><scope>test</scope></dependency><!-- 其他依赖 -->
</dependencies>

然后，创建一个 Spring Kafka 流处理应用程序：

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.serialization.Serdes;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;@SpringBootApplication
@EnableKafka
public class SpringKafkaApp {public static void main(String[] args) {SpringApplication.run(SpringKafkaApp.class, args);}// 创建输入和输出主题@Beanpublic NewTopic inputTopic() {return new NewTopic("input-topic", 1, (short) 1);}@Beanpublic NewTopic outputTopic() {return new NewTopic("output-topic", 1, (short) 1);}// 定义流处理拓扑@KafkaListener(topics = "input-topic")public void processInputMessage(@Payload String message,@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {// 在这里进行数据转换和处理操作String processedMessage = message.toUpperCase();// 发送处理结果到输出主题kafkaTemplate().send("output-topic", processedMessage);}// 创建 KafkaTemplate 实例@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}// 创建 ProducerFactory 实例@Beanpublic 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, StringSerializer.class);return new DefaultKafkaProducerFactory<>(configProps);}
}

通过 @EnableKafka 注解启用 Spring Kafka。

通过 @Bean 注解创建了输入主题和输出主题的 NewTopic 实例。

使用 @KafkaListener 注解的方法作为消息监听器，监听名为 "input-topic" 的输入主题。

在 processInputMessage 方法中，我们可以进行数据转换和处理操作。在这个示例中，我们将收到的消息转换为大写。

然后，我们使用 KafkaTemplate 将处理结果发送到名为 "output-topic" 的输出主题。

通过 @Bean 注解创建了 KafkaTemplate 和 ProducerFactory 的实例，用于发送消息到 Kafka。

本期到这啦我们下期再见~

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/513505.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

三星成功研发出业界首款12层堆叠HBM3E

三星成功研发出业界首款12层堆叠HBM3E

三星电子有限公司成功研发出业界首款12层堆叠HBM3E DRAM——HBM3E 12H&#xff0c;这是迄今为止容量最大的HBM产品。这款新型HBM3E 12H内存模块提供了高达1,280GB/s的史上最高带宽&#xff0c;并拥有36GB的存储容量&#xff0c;相较于之前的8层堆叠HBM3 8H&#xff0c;在带宽和…
阅读更多...
记录一次排查负载均衡不能创建的排查过程

记录一次排查负载均衡不能创建的排查过程

故障现象&#xff0c;某云上&#xff0c;运维同事在创建负载均衡的时候&#xff0c;发现可以创建资源&#xff0c;但是创建完之后&#xff0c;不显示对应的负载均衡。 创建负载均衡时候&#xff0c;按f12发现console有如下报错 后来请后端网络同事排查日志发现&#xff0c;是后…
阅读更多...
《PyTorch深度学习实践》第十一讲卷积神经网络进阶

《PyTorch深度学习实践》第十一讲卷积神经网络进阶

一、 1、卷积核超参数选择困难&#xff0c;自动找到卷积的最佳组合。 2、1x1卷积核&#xff0c;不同通道的信息融合。使用1x1卷积核虽然参数量增加了&#xff0c;但是能够显著的降低计算量(operations) 3、Inception Moudel由4个分支组成&#xff0c;要分清哪些是在Init里定义…
阅读更多...
汽车后视镜反射率检测仪厂家

汽车后视镜反射率检测仪厂家

随着汽车工业的快速发展&#xff0c;汽车后视镜作为驾驶员观察车辆周围环境的重要工具&#xff0c;其性能和质量对于交通安全至关重要。汽车后视镜的反射率检测仪是一种用于检测汽车后视镜反射性能的专业设备&#xff0c;其重要性不言而喻。本文将重点介绍汽车后视镜反射率检测…
阅读更多...
猜数字小游戏

猜数字小游戏

目录 java&#xff1a; c语言&#xff1a; java编写&#xff1a; 首先我们要获取随机数 java帮我们写好了一个类叫Random&#xff0c;这个类就可以生成一个随机数 那我们该如何使用Random类呢&#xff1f; 1、导包———Random这个类在哪呢&#xff08;导包必须出现在类定义…
阅读更多...
短视频矩阵系统--抖去推---年后技术还能迭代更新开发运营吗？

短视频矩阵系统--抖去推---年后技术还能迭代更新开发运营吗？

短视频矩阵系统#短视频矩阵系统已经开发3年&#xff0c;年后这个市场还能继续搞吗&#xff1f;目前市面上开发短视频账号矩阵系统的源头公司已经不多了吧&#xff0c;或者说都已经被市场被官方平台的政策影响的不做了吧&#xff0c;做了3年多的矩阵系统开发到现在真的是心里没有…
阅读更多...
二叉树——700. 二叉搜索树中的搜索、98. 验证二叉搜索树

二叉树——700. 二叉搜索树中的搜索、98. 验证二叉搜索树

二叉搜索树中的搜索 给定二叉搜索树&#xff08;BST&#xff09;的根节点 root 和一个整数值 val。 你需要在 BST 中找到节点值等于 val 的节点。 返回以该节点为根的子树。 如果节点不存在&#xff0c;则返回 null 。 示例 1: 输入&#xff1a;root [4,2,7,1,3], val 2 …
阅读更多...
2024全国水科技大会暨新能源及电子行业废水论坛（十一）

2024全国水科技大会暨新能源及电子行业废水论坛（十一）

一、会议背景 为深入学习贯彻《中共中央、国务院关于全面推进美丽中国建设的意见》&#xff0c;全面贯彻实施《固体废物污染环境防治法》、《“十四五”全国城市基础设施建设规划》&#xff0c;推进我国污泥处理工程建设&#xff0c;提高处理产物资源化利用水平&#xff0c;促进…
阅读更多...
分享两会焦点热词，深化AI多场景应用，推动大模型垂直化与产业化

分享两会焦点热词，深化AI多场景应用，推动大模型垂直化与产业化

大模型、机器人、智能制造、自动驾驶……过去一年&#xff0c;人工智能频上头条&#xff0c;也成为今年北京两会上的热词。全球新一轮技术变革加速来临&#xff0c;大模型作为人工智能发展的核心引擎&#xff0c;正引发一场全新的工业革命。 在这一关键时期&#xff0c;全国政协…
阅读更多...
7.2.2 用坐标表示平移 教案设计及课堂检测设计

7.2.2 用坐标表示平移 教案设计及课堂检测设计

【学习目标】 1&#xff0e;掌握坐标变化和图形平移的关系&#xff0c;能用点的平移规律求点平移后的点的坐标&#xff0e; 2&#xff0e;会按要求画出平移后的图形&#xff0c;并写出顶点的坐标&#xff0e;
阅读更多...
2024上教师资格证模考大赛（第六季）信息技术学科知识与教学能力

2024上教师资格证模考大赛（第六季）信息技术学科知识与教学能力

2.利用搜索引擎在网上搜索信息&#xff0c;有时需要增大搜索范围以便于准确查找需要的信息&#xff0c;下列能增大搜索范围的做法是&#xff08; B&#xff09;。 A使用逻辑控制符and&#xff0c;增加搜索关键词 B减少所用的关键词&#xff0c;减少搜索条件 C使用多个关键词&…
阅读更多...
【PCL】（十八）随机采样一致性(RANSAC)模型

【PCL】（十八）随机采样一致性(RANSAC)模型

&#xff08;十八&#xff09;RANdom SAmple Consensus(RANSAC)模型 假设我们正在查看的所有数据都由内部值和外部值(异常值)组成&#xff0c;其中内部值可以用一组特定参数值的模型来解释&#xff0c;而异常值则不适合该模型。RANSAC用于估计内部值的数学模型参数。 下面的图…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023