Kafka的安装及接入SpringBoot

环境:windows、jdk1.8、springboot2

Apache KafkaApache Kafka: A Distributed Streaming Platform.icon-default.png?t=N7T8https://kafka.apache.org/

1.概述

        Kafka 是一种高性能、分布式的消息队列系统,最初由 LinkedIn 公司开发,并于2011年成为 Apache 顶级项目。它设计用于处理大规模的实时数据流,具有高吞吐量、低延迟、持久性等特点,被广泛应用于构建实时数据管道、日志收集、事件驱动架构等场景。

        详细概述见Kafka概述:

1.1 Kafka的作用

  • 发布和订阅记录流
  • 持久存储记录流,Kafka中的数据即使消费后也不会消失
  • 在系统或应用之间构建可靠获取数据的实时流数据管道
  • 构建转换或响应数据流的实时流应用程序
  • Kafka可以处理源源不断产生的数据

1.2 Kafka的一些概念

  • topic:Kafka将消息分门别类,每一类的消息称之为一个主题(Topic 就是Rabbitmq中的queue)

  • producer:发布消息的对象称之为主题生产者(Kafka topic producer)

  • consumer:订阅消息并处理发布的消息的对象称之为主题消费者(consumers)

  • broker:已发布的消息保存在一组服务器中,称之为Kafka集群。集群中的每一个服务器都是一个代理(Broker)。 消费者可以订阅一个或多个主题(topic),并从Broker拉数据,从而消费这些已发布的消息。

2.Kafka下载安装

Apache KafkaApache Kafka: A Distributed Streaming Platform.icon-default.png?t=N7T8https://kafka.apache.org/downloads        选择最新版就可以

2.1 配置kafka

        解压下载的文件,修改 config 文件夹下的 zookeeper.properties

        修改 config 文件夹下的 server.properties

        当需要外网访问时要配置advertised.listeners(比如连云服务器的kafka)

advertised.listeners=PLAINTEXT://xxx.xxx.xxx.xxx:9092

 

2.2 启动 zookeeper

        Zookeeper 在 Kafka 中充当了分布式协调服务的角色,帮助 Kafka 实现了集群管理、元数据存储、故障恢复、领导者选举等功能,是 Kafka 高可用性、可靠性和分布式特性的重要支撑。

        kafka_2.13-3.7.0\bin\windows文件夹中输入命令:

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

        可以本地访问看一下:http://localhost:2181/ 

2.3 启动Kafka 

        kafka_2.13-3.7.0\bin\windows文件夹中输入命令:

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

        访问路径: http://localhost:9092/ 

2.4 便捷启动脚本

        两个脚本放到Kafka的目录(kafka_2.13-3.7.0)中

cd bin\windows

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

cd bin\windows

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

3.springboot集成Kafka

3.1 环境搭建

(1)添加pom依赖

<!-- 继承Spring boot工程 -->
<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.8.RELEASE</version>
</parent>
<properties><fastjson.version>1.2.58</fastjson.version>
</properties>
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- kafkfa --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency>
</dependencies>

(2)配置类application.yml

        生产者:

spring:kafka:bootstrap-servers: xxx.xxx.xxx.xxx:9092producer:retries: 0key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

        消费者:

spring:kafka:bootstrap-servers: xxx.xxx.xxx.xxx:9092consumer:group-id: kafka-demo-kafka-groupkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

(3)启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class KafkaApp {public static void main(String[] args) {SpringApplication.run(KafkaApp.class, args);}
}

3.2 消息生产者

        junit测试,新建消息发送方

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.junit4.SpringRunner;
​
​
@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaSendTest {@Autowiredprivate KafkaTemplate<String,String> kafkaTemplate; //如果这里有红色波浪线,那是假错误
​@Testpublic void sendMsg(){String topic = "spring_test";kafkaTemplate.send(topic,"hello spring boot kafka!");System.out.println("发送成功.");while (true){ //保存加载ioc容器
​}}
}

3.3 消息消费者

        新建监听类:

​
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
​
@Component
public class MyKafkaListener {
​//    以下两种方法都行// 指定监听的主题
//    @KafkaListener(topics = "spring_test")
//    public void receiveMsg(String message){
//        System.out.println("接收到的消息:"+message);
//    }
​@KafkaListener(topics = "spring_test")public void handleMessage(ConsumerRecord<String, String> record) {System.out.println("接收到消息,偏移量为: " + record.offset() + " 消息为: " + record.value());}
}

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

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

相关文章

C语言-课程管理系统-大作业

C语言编写课程管理系统 1 需求分析2 需要查的知识点3 数据结构和基础函数功能3.1 课程数据结构3.2 菜单和选择项3.3 从文件中加载课程信息到内存3.4 将内存中的课程信息保存到文件3.5 将输入的字符串格式化为课程信息结构体 4 主要功能函数4.1 录入课程信息函数4.2 浏览课程信息…

2024全新小狐狸AI免授权源码

源码安装说明&#xff1a; 下 载 地 址 &#xff1a; runruncode.com/php/19757.html 1. 在宝塔新建一个站点&#xff0c;选择 PHP 版本为 7.2、7.3 或 7.4。将压缩包上传到站点的根目录&#xff0c;并设置运行目录为 /public。 2. 导入数据库文件&#xff0c;该文件位于 …

gap意识

在学习的过程中&#xff0c;遇到一个知识点搞不懂&#xff0c;或是在工作的过程中&#xff0c;遇到一个难题解决不了&#xff0c;很多时候不是我们的能力不行或是智商不够&#xff0c;而是我们当前的认知与那个知识点或难题之间存在较大的 gap。比如&#xff0c;若没学过线性代…

最大9W升压型DCDC多串LED恒流驱动

描述 AP9234是一款由基准电压源、振荡电路、误差放大电路、相位补偿电路、电流限制电路等构成的CMOS升压型DC/DC LED驱动。由于内置了低导通电阻的增强型N沟道功率 MOSFET&#xff0c;因此适用于需要高效率、高输出电流的应用电路。另外&#xff0c;可通过在VSENSE端子连接电…

Python装饰器带括号和不带括号的理解

装饰器是 Python 中一个强大且灵活的特性&#xff0c;允许用户在不修改原有函数或类定义的基础上&#xff0c;为其增加额外功能。 今天在尝试自定义 Python 装饰器的时候遇到了一个问题&#xff0c;因为以前一直是使用装饰器&#xff0c;基本没有自定义过装饰器&#xff0c;所…

vue3点击添加小狗图片,vue3拆分脚本

我悄悄蒙上你的眼睛 模板和样式 <template><div class"XueXi_Hooks"><img v-for"(dog, index) in dog1List" :src"dog" :key"index" /><button click"addDog1">点我添加狗1</button><hr …

WPF之工具栏菜单栏功能区。

1&#xff0c;菜单栏&#xff0c;工具栏&#xff0c;状态栏。 1.1&#xff0c;Menu中可添加菜单分隔条<Separator></Separator>作为分割线&#xff0c;使用Separator可以通过改变其template来自定义&#xff0c;Separator是无焦点的&#xff0c;如果简单的在MenuIt…

JavaScript异步编程——08-Promise的链式调用【万字长文,感谢支持】

前言 实际开发中&#xff0c;我们经常需要先后请求多个接口&#xff1a;发送第一次网络请求后&#xff0c;等待请求结果&#xff1b;有结果后&#xff0c;然后发送第二次网络请求&#xff0c;等待请求结果&#xff1b;有结果后&#xff0c;然后发送第三次网络请求。以此类推。…

【ARM 嵌入式 C 入门及渐进 1.2 -- 是否为 n 字节对齐】

文章目录 是否为 n 字节对齐 是否为 n 字节对齐 在C语言中&#xff0c;你可以定义一个宏来检查一个给定地址是否是n字节对齐的。这里的n应该是2的幂&#xff08;例如&#xff0c;2、4、8、16等&#xff09;。要做到这一点&#xff0c;可以利用位操作的特性。具体地&#xff0c…

品鉴中的平衡之美:如何欣赏红酒的口感与风格和谐

品鉴云仓酒庄雷盛红酒的过程&#xff0c;是对其口感与风格和谐的追求和欣赏。平衡是红酒品质的重要标志之一&#xff0c;它体现在红酒的色泽、香气、口感和余味等多个方面。通过欣赏红酒的平衡之美&#xff0c;我们可以更好地领略其精妙之处&#xff0c;感受其带来的美妙滋味。…

【Unity Animation 2D】Unity Animation 2D骨骼绑定与动画制作

一、图片格式为png格式&#xff0c;并且角色各部分分离 图片参数设置 需要将Sprite Mode设置为Single&#xff0c;否则图片不能作为一个整体 1、创建骨骼 1.1 旋转Create Bone&#xff0c;点击鼠标左键确定骨骼位置&#xff0c;移动鼠标再次点击鼠标左键确定骨骼&#xff0c…

大数据面试题第一期*4

题1、HDFS存储机制 &#xff08;1&#xff09;客户端向namenode请求上传文件 &#xff0c;namenode检查目标文件是否已存在 &#xff0c;父目录是否存在。 &#xff08;2&#xff09;namenode返回是否可以上传。 &#xff08;3&#xff09;客户端请求第一个 block上传到哪几个d…

Docker in Docker 的原理与实践

概述 Docker in Docker&#xff08;DinD&#xff09;是一个让 Docker 容器内可以运行另一个 Docker 沙箱环境的技术。常用于持续集成&#xff08;CI&#xff09;工作流程&#xff0c;其中需要构建和推送 Docker 镜像&#xff0c;而不污染主宿主机的 Docker 环境。 Docker in …

JS_jq选择器合集

基本选择器 选择器返回示范描述#id单个元素$("#test").class 集合元素 $(".test")element集合元素$("p") 所有p原素*集合元素$("*")所有原素select1,select2,selectn集合元素$("div,span,p.myClass")所有匹配原素 层次选择…

嵌入式C语言高级教程:实现基于STM32的智能健康监测手环

智能健康监测手环能够实时监控用户的生理参数&#xff0c;如心率、体温和活动量&#xff0c;对于健康管理和疾病预防非常有帮助。本教程将指导您如何在STM32微控制器上实现一个基本的智能健康监测手环。 一、开发环境准备 硬件要求 微控制器&#xff1a;STM32L476RG&#xf…

RS3236-3.3YUTDN4功能和参数介绍及PDF资料

RS3236-3.3YUTDN4功能和参数介绍及PDF资料-公司新闻-配芯易-深圳市亚泰盈科电子有限公司 品牌: RUNIC(润石) 封装: XDFN-4-EP(1x1) 描述: 带过温保护 输出类型: 固定 最大输入电压: 7.5V 输出电压: 3.3V 最大输出电流: 500mA RS3236-3.3YUTDN4 是一款低压差线性稳压器&#x…

kkfileview部署踩坑记录——kkfile部署启动失败、预览出错、乱码问题的处理

预览失败&#xff1a;报错 org.jodconverter.core.office.OfficeException: Could not store document: testdoc.pdf java.lang.NullPointerException: Could not open document: 1691539546735.docx 错误原因 由于kkfileView在linux上默认使用openOffice来实现转换 解决方…

leetcode 2316.统计无向图中无法互相到达点对数

思路&#xff1a;并查集 其实就是连通块的一个变形题目&#xff0c;一般的连通块题目要我们求的是连通个数&#xff0c;或者能不能到达&#xff0c;这里反过来问了。 首先&#xff0c;我们用dfs也是可以做到的&#xff0c;在dfs中统计每一个连通块的个数&#xff0c;然后用乘…

Spring原理分析--获取Environment资源对象

1.使用getEnvironment()获取环境信息 ApplicationContext接口继承了EnvironmentCapable接口&#xff0c;可以通过getEnvironment()获取Environment配置信息&#xff0c;例如&#xff1a; SpringBootApplication public class A01 {public static void main(String[] args) th…

解决方案:对数据进行负采样随机抽取1000W,用Pandas如何实现

文章目录 一、现象二、解决方案 一、现象 做建模的时候&#xff0c;有时候需要对数据进行负采样&#xff0c;就需要随机抽取数据&#xff0c;之前用SQL实现过order by rand()&#xff0c;附上链接解决方案&#xff1a;用户号出现多行&#xff0c;如何从中取其一并随机抽取100个…