首页 > 其他分享 >消息队列kafka中间件详解：案例解析(第10天)

消息队列kafka中间件详解：案例解析(第10天)

时间：2024-06-21 19:29:02浏览次数：10  
标签：10 -- bootstrap 中间件 server topic kafka Kafka

系列文章目录

  • 1- 消息队列（熟悉）
  • 2- Kafka的基本介绍（掌握架构，其他了解）
  • 3- Kafka的相关使用（掌握kafka常用shell命令）
  • 4- Kafka的Python API的操作（熟悉）

文章目录

前言

本文主讲述了 消息队列，Kafka的架构，Kafka的相关使用和常用shell命令，Kafka的Python API的操作；

一、消息队列（熟悉）

1、产生背景

消息队列：指的数据在一个容器中，从容器中一端传递到另一端的过程

消息(message): 指的是数据，只不过这个数据存在一定流动状态
队列(queue): 指的容器，可以存储数据，只不过这个容器具备FIFO(先进先出)特性

思考: 公共容器需要具备什么特点?
1- 公共性: 各个程序都可以与之对接
2- FIFO特性: 先进先出
3- 具备高效的并发能力: 能够承载海量数据
4- 具备一定的容错能力: 比如支持重新读取消息方案

2、消息队列介绍

2.1 常见的消息队列产品

MQ：message queue消息队列

activeMQ: 出现时期比较早的一款消息队列的中间件产品，在早期使用人群是非常多，目前整个社区活跃度严重下降，使用人群很少了
rabbitMQ: 此款是目前使用人群比较多的一款消息队列的中间件的产品，社区活跃度比较高，主要是应用传统业务领域中
rocketMQ: 是阿里推出的一款消息队列的中间件的产品，目前主要是在阿里系环境中使用，目前支持的客户端比较少，主要是Java中应用较多
Kafka: Apache旗下的顶级开源项目，是一款消息队列的中间件产品项目来源于领英，是大数据体系中目前为止最为常用的一款消息队列的产品

在这里插入图片描述

2.2 应用场景
  • 应用解耦合
  • 异步处理
  • 限流削峰
  • 消息驱动系统
2.3 消息队列中两种消息模型
在Java中, 为了能够集成消息队列的产品, 专门提供了一个消息队列的协议: JMS(Java Message Server)  java消息服务

消息队列中两个角色: 生产者(producer) 和 消费者(consumer)
生产者: 生产/发送消息到消息队列中
消费者: 从消息队列中获取消息

在JMS规范中, 专门规定了两种消息消费模型: 
1- 点对点消费模型: 指的一条消息最终只能被一个消费者所消费。微信聊天的私聊
2- 发布订阅消费模型: 指的一条消息最终被多个消费者所消费。微信聊天的群聊

二、Kafka的基本介绍

1、Kafka基本介绍

​ Kafka是一款消息队列的中间件产品, 来源于领英公司, 后期贡献给了Apache, 目前是Aapche旗下的顶级开源项目, 采用语言是Scala

​ 官方地址: http://kafka.apache.org

​

kafka的特点:

  • 可靠性：Kafka集群是分布式的，并且有多副本的机制。数据可以自动复制
  • 可扩展性：Kafka集群可以灵活的调整，在线扩容
  • 耐用性：Kafka数据保存在磁盘上面，数据并且有多副本的机制。数据持久化，而且可以一定程度上防止数据丢失
  • 高性能：Kafka可以存储海量的数据，虽然是使用磁盘进行数据存储，但是Kafka有各种优化手段（例如：磁盘的顺序读写、零拷贝等）提高数据的读写速度（吞吐量）

2、回顾zookeeper知识

Kafka需要使用到zookeeper服务!

  • 启动zookeeper服务
# 三台都需要启动zookeeper服务
[root@node1 ~]# /export/server/zookeeper/bin/zkServer.sh start
[root@node2 ~]# /export/server/zookeeper/bin/zkServer.sh start
[root@node3 ~]# /export/server/zookeeper/bin/zkServer.sh start
  • zookeeper工具连接

把ZooInspector.rar解压然后进入ZooInspector\build双击zookeeper-dev-ZooInspector.jar（资源已经上传博客）

在这里插入图片描述

3、Kafka的架构（掌握）

HDFS写入过程回顾：

在这里插入图片描述

Kafka架构：

在这里插入图片描述

1- Kafka中集群节点叫broker，节点和节点之间没有主从之分，地位是完全一样
2- Topic：主题/话题，是业务层面对消息进行分类的。
3- 一个Topic可以设置多个Partition分区。
4- 同一个Partition分区可以设置多个副本，但是副本数不能超过（>）集群broker节点的个数
5- 虽然broker节点间没有主从之分，但是同一个Partition分区的不同副本间有主从之分，分为了Leader主副本和Follower从副本
6- 生产者将数据首先发送给到Leader主副本，接着是Leader主副本主动的往Follower从副本上同步消息
7- Zookeeper用来管理集群，以及管理元数据信息
8- ISR同步列表。该列表中存放的是与Leader主副本消息同步程度最接近的Follower从副本，也就是消息最小的一个列表。该列表作用，当Leader主副本无法对外提供服务的时候，会从该ISR列表中选择一个Follower从副本变成Leader主副本，对外提供服务


相关名词:
Kafka Cluster: Kafka集群
Topic: 主题/话题
Broker: Kafka中的节点
Producer: 生产者，负责生产/发送消息到Kafka中
Consumer: 消费者，负责从Kafka中获取消息
Partition: 分区。一个Topic可以设置多个分区，没有数量限制

三、Kafka的shell命令使用（掌握）

​ Kafka本质上就是一个消息队列的中间件的产品，主要负责消息数据的传递。也就说学习Kafka 也就是学习如何使用Kafka生产数据，以及如何使用Kafka来消费数据

1、topics操作

注意:

创建topic不指定分区数和副本数,默认都是1个

分区数可以后期通过alter增大,但是不能减小

副本数一旦确定,不能修改!

参数如下:

cd /export/server/kafka/bin

./kafka-topics.sh 参数说明:
	--bootstrap-server: Kafka集群中broker服务器
	--topic: 指定Topic名称
	--partitions: 设置Topic的分区数,可以省略不写
	--replication-factor: 设置Topic分区的副本数,可以省略不写
	
	--create: 指定操作类型。这里是新建Topic
	--delete: 指定操作类型。这里是删除Topic
	--alter: 指定操作类型。这里是修改Topic
	--list: 指定操作类型。这里是查看所有Topic列表
	--describe: 指定操作类型。这里是查看详细且具体的Topic信息
  • 1- 创建Topic
# 创建topic,默认1个分区,1个副本
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic itcast 

# 注意: 如果副本数超过了集群broker节点个数，就会报错
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic itheima --partitions 4 --replication-factor 4

在这里插入图片描述

# 把replication-factor改成3以内就能创建成功了
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic itheima --partitions 4 --replication-factor 3
  • 2- 查看Topic
# --list查看所有topic
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --list

# --describe 可以查看详细Topic信息
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --describe 

# --describe 可以查看具体Topic信息
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --describe --topic itheima

在这里插入图片描述

当然也可使用zookeeper客户端查看

在这里插入图片描述

  • 3- 修改Topic
# 增大topic分区
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --alter --topic itcast --partitions 4

# 注意: partitions分区,只能增大，不能减小。而且没有数量限制
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --alter --topic itcast --partitions 1

在这里插入图片描述

# 注意: 副本既不能增大，也不能减小
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --alter --topic itcast --partitions 4 --replication-factor 2

在这里插入图片描述

  • 4- 删除Topic
# 再创建一个spark主题
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic spark

/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --list

# 删除spark主题

/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --delete --topic spark

/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --list

2、producer和consumer操作

消费者要和生产者指定是同一个topic主题,才能接收到消息

参数如下:

cd /export/server/kafka/bin

./kafka-console-producer.sh 参数说明
	--broker-list: Kafka集群中broker服务器
	--topic: 指定Topic
	
./kafka-console-consumer.sh 参数说明
	--bootstrap-server: Kafka集群中broker连接信息
	--topic: 指定Topic
	latest: 消费者（默认）从最新的地方开始消费
	--from-beginning: 指定该参数以后，会从最旧的地方开始消费
	--max-messages: 最多消费的条数。
  • 1- 模拟生产者Producer
# 为了方便演示再创建一个spark
/export/server/kafka/bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic spark

# 模拟生产者给spark发送消息
/export/server/kafka/bin/kafka-console-producer.sh --broker-list node1:9092 --topic spark
  • 2- 模拟消费者Consumer
# 模拟消费者从spark获取消息,默认每次拿最新的
/export/server/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic spark 

# --from-beginning 会从最旧的地方开始消费
/export/server/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic spark --from-beginning

# --max-messages x 可以设置从最旧的地方最大消费次数x
/export/server/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic spark --from-beginning --max-messages 5

注意：

我们有时候发现消费者打印出来的消息和生产者生产的顺序不一致，是乱序的。原因如下：

topic有多个分区，底层是多线程来读取数据并进行打印输出。因此会存在乱序现象

3、bootstrap-server和zookeeper以及broker-list的区别:

旧版（<v2.2）: kafka-topics.sh --zookeeper node1:2181,node2:2181,node3:2181/kafka --create --topic ..
注意: 旧版用--zookeeper参数，主机名（或IP）和端口用ZooKeeper的2181，也就是server.properties文件中zookeeper.connect属性的配置值.

新版(>v2.2): kafka-topics.sh --bootstrap-server node1:9092 --create --topic ..
注意: 新版用--bootstrap-server参数，主机名（或IP）和端口用某个节点的即可，即主机名（或主机IP）:9092。9092是Kafka的监听端口



broker-list:broker指的是kafka的服务端，可以是一个服务器也可以是一个集群。producer和consumer都相当于这个服务端的客户端。一般我们再使用console producer的时候，broker-list参数是必备参数，另外一个必备的参数是topic

bootstrap-servers: 指的是kafka集群的服务器地址，这个和broker-list功能是一样的，只不过我们在console producer要求用broker-list,其他地方都采用bootstrap-servers。

四、kafka tools工具使用（熟悉）

可以在可视化的工具通过点击来操作kafka完成主题的创建，分区等操作，资源包已经上传到博客第10天内

在这里插入图片描述

注意: 安装完后桌面不会有快捷方式,需要去电脑上搜索,或者去自己选的安装位置找到发送快捷方式到桌面!

在这里插入图片描述

在这里插入图片描述

1、连接配置

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

2、创建主题

在这里插入图片描述

在这里插入图片描述

3、删除主题

在这里插入图片描述

在这里插入图片描述

4、主题下的数据查看

在这里插入图片描述

5、数据显示问题说明

在这里插入图片描述

  • 修改工具的数据显示类型

    在这里插入图片描述

    在这里插入图片描述

​ 在这里插入图片描述

6、发送消息数据到kafka

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

五、Kafka的Python API的操作（熟悉）

1、模块安装

纯Python的方式操作Kafka。

准备工作：在node1的节点上安装一个python用于操作Kafka的库

安装kafka-python 模模块 ,模块中提供了操作kafka的方法

在线安装

在node1上安装就可以，需要保证服务器能够连接网络

安装命令: python -m pip install kafka-python -i https://pypi.tuna.tsinghua.edu.cn/simple

在这里插入图片描述

离线安装

将kafka_python-2.0.2-py2.py3-none-any.whl安装包上传服务器software目录下进行安装

安装命令: pip install kafka_python-2.0.2-py2.py3-none-any.whl
2、模块使用

API使用的参考文档: https://kafka-python.readthedocs.io/en/master/usage.html#kafkaproducer

模块中封装了两个类，

一个是生成者类KafkaProducer，提供了向kafka写数据的方法

另一个是消费者类KafkaConsumer，提供了读取kafka数据的方法

2.1 完成生产者代码

生成者类KafkaProducer，提供了向kafka写数据的方法

send(topic,valu)方法: 发送消息
topic参数:指定向哪个主题发送消息
value参数:指定发送的消息数据 ，数据类型要求是bytes类型

示例:

# 导包
from kafka import KafkaProducer

# 编写代码
if __name__ == '__main__':
    # 创建生产者对象并指定对应服务器
    producer = KafkaProducer(bootstrap_servers=['node1:9092'])
    # 发送消息
    for i in range(1,101):
        future = producer.send('kafka', f'hi_kafka_{i}'.encode())
        # 获取元数据
        record_metadata = future.get()
        # 从元数据中获取主题,分区,偏移
        print(record_metadata.topic)
        print(record_metadata.partition)
        print(record_metadata.offset)
2.2 完成消费者代码

消费者类KafkaConsumer，提供了读取kafka数据的方法

KafkaConsumer(topic,bootstrap_servers)
第一个参数:指定消费者连接的主题，
第二个参数:指定消费者连接的kafka服务器

示例:

# 导包
from kafka import KafkaConsumer

# 编写代码
if __name__ == '__main__':

    # 创建消费者对象
    consumer = KafkaConsumer('kafka',bootstrap_servers=['node1:9092'])
    # 遍历对象
    for message in consumer:

        # 格式化打印,设置相关参数
        # 因为value是二进制,需要decode解码
        print ("主题:%s,分区:%d,偏移:%d : key=%s value=%s"
               % (message.topic, message.partition,message.offset, message.key, message.value.decode('utf8')))

可能遇到的错误：

在这里插入图片描述

原因: 服务器环境有问题。是因为服务器上既安装了kafka-python的第三方依赖，同时还安装kafka的第三方依赖。可以通过pip list | grep kafka进行确定
解决办法: 先将这两个第三方依赖全部卸载，然后再重新执行如下命令
python -m pip install kafka-python -i https://pypi.tuna.tsinghua.edu.cn/simple

标签：10,--,bootstrap,中间件,server,topic,kafka,Kafka
From： https://blog.csdn.net/syhiiu/article/details/139856572

相关文章

  • ch10 降维与度量学习
    降维的动机从k-近邻算法的角度看降维如果给定测试样本\(x\)与最近邻样本\(z\)，那么正确率就为\[P(acc)=P(c_1=c_2|x,z)=\sum{c\in\mathcal{C}}P(c_1=c_2=c|x,z)=\sum_{c\in\mathcal{C}}P(c_1=c|x)P(c_2=c|z)\]如果在度量空间中满足密采样假......
  • 游泳耳机品牌排行榜，10大实力超群的游泳耳机分享！
    在当今快节奏的生活中，运动已成为许多人不可或缺的一部分，不仅为了健康，也是释放压力、提升生活品质的有效方式。而随着科技与健身的深度融合，智能穿戴设备尤其是专为运动设计的耳机，正逐渐成为运动爱好者的新宠。对于热爱水上运动，特别是游泳的人群而言，一款性能卓越的游泳耳机，不仅能......
  • CF1083F The Fair Nut and Amusing Xor
    题意给定两个长度为\(n\)的数列\(a,b\)，规定每次操作为选取一段长度为\(k\)的子段异或上任意自然数。对于每次查询，先单点修改\(a\)或\(b\)，你需要求出最小的操作次数，或无法使得\(a\)在若干次操作后变为\(b\)。\(n\le2\times10^5\)Sol差个分先，区间异或变为\(......
  • Windows 10 version 22H2 (updated Jun 2024) 中文版、英文版下载
    Windows10version22H2(updatedJun2024)中文版、英文版下载Windows1022H2企业版arm64x64请访问原文链接：https://sysin.org/blog/windows-10/，查看最新版。原创作品，转载请保留出处。作者主页：sysin.org直接上链接，详细说明请访问原文查看。下载地址语言：简体中文、......
  • AP2400 5-150V 10-6000mA 耐高温 高精度 降压恒流LED驱动芯片 手电筒电动车照明灯
    产品描述AP2400是一款PWM工作模式,高效率、外围简单、外驱功率管，适用于输入的高精度降压LED恒流驱动芯片。外驱MOS，最大输出电流可达6A。AP2400可实现三段功能切换，通过MODE1/2/3切换三种功能模式：全亮，半亮，爆闪AP2400工作频率固定在150KHZ左右，同时内置抖频电路，可以......
  • AP5161恒流驱动芯片 输入100V 输出3A 内置MOS 汽车灯方案
    概述AP5161是一款高精度降压型大功率LED恒流驱动芯片。适用于输入电压100V以内的大功率LED恒流驱动电源。专利的高端电流检测、固定频率、电流模PWM控制方式，具有优异的线性调整率和负载调整率。芯片采用的特有恒流控制方式，使得LED输出电流精度达到±3%以内。芯......
  • AP5127 DC-DC降压恒流IC 输入12-100V 输出2.5A LED车灯方案
    产品描述AP5127是一款PWM工作模式,高效率、外围简单、内置功率管，适用于12-100V输入的高精度降压LED恒流驱动芯片。输出最大功率可达25W，最大电流2.5A。AP5127可实现全亮/半亮功能切换，通过MODE切换：全亮/半亮/循环模式。AP5127工作频率固定在140KHZ，同时内置抖频......
  • 【漏洞挖掘】——101、搜索型注入深入刨析
    基本介绍网站为了用户便于搜索网站中的资源往往会加入搜索功能，如果底层的代码忽略了对搜索变量的过滤，则有可能会造成了搜索型注入漏洞(又称文本框注入)，对于常见的搜索框而言，后台常用的SQL语句如下所示，在这里参数如果未经过滤直接拼接到SQL语句中则存在注入，注入方法类似于字符......
  • 【漏洞挖掘】——102、Blind注入类型深入刨析(上)
    基本介绍BlindInjection(即盲注)，当SQL语句执行成功或失败不会给出明确的回显信息或者只有一个提示类的信息，类似于true\false时，此时可以考虑盲注，常见的盲注有以下几种：基于布尔的盲注基于时间的盲注布尔盲注适用场景适用于对于sql语句执行成功或失败各有一个提示结果，例......
  • 【漏洞挖掘】——103、Blind注入类型深入刨析(中)
    时间盲注使用场景没有True/False，也没有任何错误提示类信息，在这种情况下可以考虑通过时间线的长短来判断是否存在SQL注入，即基于时间的盲注！函数简介sleep()：用于延迟IF(Condition,A,B)：当Condition为TRUE时，返回A；当Condition为FALSE时，返回BBENCHMARK(count,expr)：重复count......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99