河南网站建设及推广,thinkcmf跟wordpress,建筑建材网站设计费用,电脑制作网站用哪个软件系列文章目录
上手第一关#xff0c;手把手教你安装kafka与可视化工具kafka-eagle Kafka是什么#xff0c;以及如何使用SpringBoot对接Kafka 系列文章目录一、Kafka与流处理二、Spring Boot与Kafka的整合Demo1. 新建springboot工程2. 添加Kafka依赖3. 配置Kafka4. 创建Kafka…系列文章目录
上手第一关手把手教你安装kafka与可视化工具kafka-eagle Kafka是什么以及如何使用SpringBoot对接Kafka 系列文章目录一、Kafka与流处理二、Spring Boot与Kafka的整合Demo1. 新建springboot工程2. 添加Kafka依赖3. 配置Kafka4. 创建Kafka生产者5. 创建Kafka消费者6. 应用程序入口 三、启动与验证四、KafkaTemplate 介绍总结 继上一次教大家手把手安装kafka后今天我们直接来到入门实操教程也就是使用SpringBoot该怎么对接和使用kafka。当然在一开始我们也会比较细致的介绍一下kafka本身。那么话不多说马上开始今天的学习吧 作者简介战斧从事金融IT行业有着多年一线开发、架构经验爱好广泛乐于分享致力于创作更多高质量内容 本文收录于 kafka 专栏有需要者可直接订阅专栏实时获取更新 高质量专栏 云原生、RabbitMQ、Spring全家桶 等仍在更新欢迎指导 Zookeeper Redis dubbo docker netty等诸多框架以及架构与分布式专题即将上线敬请期待 一、Kafka与流处理
我们先来看看比较正式的介绍Kafka是一种流处理平台由LinkedIn公司创建现在是Apache下的开源项目。Kafka通过发布/订阅机制实现消息的异步传输和处理。它具有高吞吐量、低延迟、可伸缩性和可靠性等优点使其成为了流处理和实时数据管道的首选解决方案
介绍其实是比较清晰的如果你是第一次接触“流处理”概念我们也可以做一点解释流处理指的是对连续、实时产生的数据流进行实时处理、计算和分析的过程。 假设你正在玩一款在线游戏其他玩家的动作和游戏事件会实时地传到服务器上。这些事件就形成了一条数据流。在流处理中我们会对这条数据流进行实时处理例如计算每个玩家的分数、监控游戏区域内的异常情况、统计玩家在线时长等等。这样游戏管理员就可以实时地监控和管理游戏而不需要等到游戏结束才进行操作。 类似的流处理还可以应用在其他实时性要求比较高的场景中例如金融交易、物联网、实时监测等。通过对数据流进行实时处理我们可以更加精准地掌握数据变化的情况并及时做出反应和调整 二、Spring Boot与Kafka的整合Demo
1. 新建springboot工程
如果你没有现成的Spring boot项目那么我们可以使用IDEA自带的Spring Initializr 来创建一个spring-boot的项目 此时我们可以直接选择使用Apache Kafka另外项目还可以加个Spring Web准备让前台调用 2. 添加Kafka依赖
如果你不是像上述一样新建的项目那你也可以选择在已有的Spring Boot应用程序中使用Kafka那么你需要在pom.xml文件中添加以下依赖
dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactIdversion2.8.11/version
/dependency3. 配置Kafka
在application.properties文件中添加以下配置
spring.kafka.bootstrap-serverslocalhost:9092
spring.kafka.consumer.group-idtest_group这里我们指定了Kafka服务器的地址和端口并配置了消费者组的ID关于消费者组的概念其实就是某一些消费者具备相同的功能因此会把他们设为同一个消费者组这样他们就不会重复消费同一条消息了。更具体地原理我们会在之后地篇章中介绍。
4. 创建Kafka生产者
在Kafka中生产者是发送消息的应用程序或服务。在Spring Boot中我们可以使用KafkaTemplate类来创建Kafka生产者
package com.zhanfu.kafkademo.service;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;Service
public class KafkaService {Autowiredprivate KafkaTemplateString, String kafkaTemplate;public void sendMessage(String message) {kafkaTemplate.send(test_topic, message);}
}这里我们使用Autowired注解来自动注入KafkaTemplate并使用send方法将消息发送到名为“test_topic”的Kafka主题中。 5. 创建Kafka消费者
在Kafka中消费者是接收并处理订阅主题消息的应用程序或服务。在Spring Boot中我们可以使用KafkaListener注解来创建Kafka消费者。
package com.zhanfu.kafkademo.listener;import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;Component
public class KafkaLis {KafkaListener(topics test_topic, groupId test_group)public void receiveMessage(String message) {System.out.println(Received message: message);}
}
6. 应用程序入口
现在我们已经完成了Spring Boot和Kafka的整合。我们可以启动Spring Boot应用程序然后发送消息并消费它以测试我们的应用程序是否正确地与Kafka集成。
package com.zhanfu.kafkademo.controller;import com.zhanfu.kafkademo.service.KafkaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;RestController
public class MessageController {Autowiredprivate KafkaService kafkaService;GetMapping(/send/{message})public String sendMessage(PathVariable String message) {kafkaService.sendMessage(message);return Message sent successfully;}
}
在这个例子中我们使用Autowired注解来自动注入KafkaProducer并通过发送消息的方法来调用sendMessage方法。最终项目整体框架如图 三、启动与验证
首先自然是启动 Kafka 怎么启动可参考 《上手第一关手把手教你安装kafka与可视化工具kafka-eagle》然后是启动我们的Spring Boot项目 然后在浏览器中输入
http://127.0.0.1:8080/send/hello最后检查我们的项目日志 可以看到整个发送和接收的流程都走通了
四、KafkaTemplate 介绍
不难看出在Springboot中使用kafka的关键在于 KafkaTemplate 它是 Spring 提供的 Kafka 生产者模版用于向 Kafka 集群发送消息。并且把 Kafka 的生产者客户端封装成了一个 Spring Bean提供更加方便易用的 API。
它有三个主要属性
producerFactory生产者工厂类用于创建 KafkaProducer 实例。defaultTopic默认主题名称如果在发送消息时没有指定主题名称则使用该默认主题。messageConverter消息转换器用于将消息对象转换为 Kafka ProducerRecord
它的主要方法
send(ProducerRecordK,V record)向指定的 Kafka 主题发送一条消息。ProducerRecord 包含了主题名称、分区编号、Key 和 Value 等信息。send(String topic, V data)向指定的 Kafka 主题发送一条消息。send(String topic, K key, V data)向指定的 Kafka 主题发送一条消息并指定消息的 Key。execute(ProducerCallbackK,V callback)使用回调方式发送消息可以自定义消息的创建过程和错误处理过程。inTransaction()启用事务多个 send 方法调用将被包装在一个事务中保证 Kafka 事务的原子性。
除了上述方法外KafkaTemplate 还提供了其他方法如 sendDefault()、sendOffsetsToTransaction() 等可以根据实际需要进行选择和使用。
需要注意的是在使用 KafkaTemplate 发送消息时应该注意消息的序列化方式、主题和分区的选择以及错误处理等问题以保证消息的可靠性和正确性。
当然很多同学可能还注意到一个细节我们在上面的Demo中我们直接将其 Autowired进我们的代码中这是怎么做到的呢换句话说这个 KafkaTemplate 为什么自己就会被spring 容器管理的呢其实这得益于SpringBoot中对Kafka有了很多自动配置的内容。如下 如上图相信对Spring Boot熟悉的同学看到 ConditionalOnClass 、 ConditionalOnMissingBean 应该就明白了。其实Spring Boot 早就贴心的为我们预留了这些自动配置只要我们引入了 spring-kafka 包使得项目中出现了 KafkaTemplate 类那么它就能被自动配置并存入Spring 容器内
总结
今天我们通过一个Demo讲解了在SpringBoot中如何对接Kafka也介绍了下关键类 KafkaTemplate 得益于Spring Boot 的自动配置开发者要做的配置内容其实并不多使用也主要是依赖其提供的API相对简单相信大家很容易也都学会了那么在后面的过程中我们将继续学习其使用并且会着重讲解 Kafka 的原理与结构
