Kafka基本介绍

消息队列

产生背景

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

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

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

消息队列介绍

常见的消息队列产品

MQ：message queue消息队列

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

应用场景

  • 应用解耦合
  • 异步处理
  • 限流削峰
  • 消息驱动系统

消息队列中两种消息模型

在Java中, 为了能够集成消息队列的产品, 专门提供了一个消息队列的协议: JMS(Java Message Server)  java消息服务

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

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

Kafka基本介绍

基本介绍

Kafka是一款消息队列的中间件产品, 来源于领英公司, 后期贡献给了Apache, 目前是Aapche旗下的顶级开源项目, 采用语言是Scala
官方地址: http://kafka.apache.org

kafka特点:

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

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的相关使用

Kafka的shell命令使用

Kafka本质上就是一个消息队列的中间件的产品，主要负责消息数据的传递。

1- 创建Topic

./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --create --topic test02 --partitions 4 --replication-factor 2  

参数说明:
	--bootstrap-server: Kafka集群中broker连接信息
	--create: 指定操作类型。这里是新建Topic
	--topic: 指定要新建的Topic名称
	--partitions: 设置Topic的分区数
	--replication-factor: 设置Topic分区的副本数

设置副本数超过了集群broker节点个数

./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --create --topic test04 --partitions 4 --replication-factor 4

在这里插入图片描述
2- 查看Topic

./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --list

参数说明:
	--bootstrap-server: Kafka集群中broker连接信息
	--list: 指定操作类型。这里是查看Kafka集群上所有可用的Topic列表

在这里插入图片描述
3- 查看具体Topic

./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --describe --topic test04
参数说明:
	--bootstrap-server: Kafka集群中broker连接信息
	--describe: 指定操作类型。这里是查看具体Topic信息

在这里插入图片描述
4- 模拟生产者Producer

./kafka-console-producer.sh --broker-list node1.itcast.cn:9092,node2.itcast.cn:9092 --topic test04
参数说明:
	--broker-list: Kafka集群中broker连接信息
	--topic: 指定要将消息发送到哪个具体的Topic

在这里插入图片描述
5- 模拟消费者Consumer

./kafka-console-consumer.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --topic test04

参数说明:
	--bootstrap-server: Kafka集群中broker连接信息
	--topic: 指定要从哪个Topic中消费消息
	--from-beginning: 指定该参数以后，会从最旧的地方开始消费
	latest: 消费者（默认）从最新的地方开始消费
	--max-messages: 最多消费的条数。满足条数后，就会自动结束
	--group: 指定消费组名称。一个消费者只能属于一个消费组；一个消费组里面可以有多个消费者。同一个Topic中的同一条数据，只能被同一个消费组中的一个消费者所消费
	
参数一般如何使用？
答: 推荐latest、--max-messages、--group一同使用。因为实际中Topic的数据量是特别大的，消费、打印都需要消耗服务器的资源，如果不限定消费的最大条数，可能造成服务器宕机。

在这里插入图片描述
6- 修改Topic

./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --alter --topic test01 --partitions 10

分区: 只能增大，不能减小。而且没有数量限制
副本: 既不能增大，也不能减小

减小分区：
./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --alter --topic test01 --partitions 1

在这里插入图片描述

修改副本数：
./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --alter --topic test01 --replication-factor 2 --partitions 11

在这里插入图片描述
7- 删除Topic

./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --delete --topic test01

参数说明:
	--bootstrap-server: Kafka集群中broker连接信息
	--delete: 指定操作类型。这里是删除Topic
	--topic: 指定要删除哪个Topic

8- 查看消费组中有多少个消费者

./kafka-consumer-groups.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --group g_01 --members --describe

在这里插入图片描述

Kafka基准测试

Kafka的基准测试, 主要是用于测试Kafka集群的吞吐量, 每秒钟最大可以生产多少条数据, 以及每秒钟最大可以消费多少条数据

测试生产的效率

1- 创建Topic

./kafka-topics.sh --create --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092 --partitions 3 --replication-factor 2 --topic test01

2- 执行生产测试命令: 测试后，会增加4GB磁盘占用

./kafka-producer-perf-test.sh --topic test01 --num-records 4000000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092 acks=1

在这里插入图片描述
3- 测试结果
在这里插入图片描述

测试消费的效率

1- 执行消费测试命令

./kafka-consumer-perf-test.sh --broker-list node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092 --topic test01 --fetch-size 1048576 --messages 500000

在这里插入图片描述
2- 测试结果:
在这里插入图片描述

Kafka的Python API的操作

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

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

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

在这里插入图片描述

完成生产者代码
import time

from kafka import KafkaProducer

# 同步发送
def sync_send():
    global topic, partition, offset
    # 2.1- 同步发送数据/消息
    metadata = producer.send("test01", value=f"hello_java_{i}".encode("UTF-8")).get()
    # metadata = producer.send("test03",value=f"hello_spark_{i}".encode("UTF-8")).get()
    # 2.2- 获取元信息中的内容
    topic = metadata.topic
    partition = metadata.partition
    """
        offset消息偏移量，从0开始编号。也就是一条消息在分区中的序号/索引
        在不同分区间，消息偏移量是无序
        在同一个分区里面，消息偏移量是有序
    """
    offset = metadata.offset
    print(f"{topic},{partition},{offset},{metadata}")


if __name__ == '__main__':

    # 1- 创建生产者
    producer = KafkaProducer(
        bootstrap_servers=["node1.itcast.cn:9092","node2.itcast.cn:9092"]
    )

    # 2- 发送消息
    for i in range(10):
        # 同步发送
        # sync_send()

        # 2.3- 异步发送
        """
            异步发送，需要等待一下，或者明确关闭Producer生产者
        """
        producer.send("test01", value=f"hello_hive_{i}".encode("UTF-8"))

    time.sleep(1)

    # 3- 释放资源/关闭生产者
    # producer.close()

在这里插入图片描述
可能遇到的错误：
在这里插入图片描述

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

if __name__ == '__main__':

    # 1- 创建消费者
    consumer = KafkaConsumer(
        "test01",
        bootstrap_servers=["node1.itcast.cn:9092", "node2.itcast.cn:9092"]
    )

    # 2- 消费消息
    for msg in consumer:
        topic = msg.topic
        partition = msg.partition
        offset = msg.offset
        # key和value消费出来都是bytes数据类型，需要进行解码
        key = msg.key
        value = msg.value

        print(f"{topic}，{partition}，{offset}，{key}，{value.decode('UTF-8')}，{msg}")

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Linux技术，winSCP连接服务器超时故障解决方案

Linux技术，winSCP连接服务器超时故障解决方案

知识改变命运，技术就是要分享，有问题随时联系，免费答疑，欢迎联系！ 故障现象 使用 sftp 协议连接主机时, 明显感觉缓慢且卡顿,并且时常出现如下报错: 点击重新连接后,又有概率重新连接上; 总之在"连接上"和&…
阅读更多...
蓝桥杯练习题（六）

蓝桥杯练习题（六）

📑前言 本文主要是【算法】——蓝桥杯练习题（六）的文章，如果有什么需要改进的地方还请大佬指出⛺️ 🎬作者简介：大家好，我是听风与他🥇 ☁️博客首页：CSDN主页听风与他 …
阅读更多...
分布式系统的三字真经CAP

分布式系统的三字真经CAP

文章目录 前言C（Consistency 数据一致性）A（Availability 服务可用性）P（Partition Tolerance 分区容错性）CAP理论最后 前言 你好，我是醉墨居士，我一起探索一下分布式系统的三字真经C…
阅读更多...
Linux完全卸载Anaconda3和MiniConda3

Linux完全卸载Anaconda3和MiniConda3

如何安装Anaconda3和MiniConda3请看这篇文章： 安装Anaconda3和MiniConda3_minianaconda3-CSDN博客文章浏览阅读474次。MiniConda3官方版是一款优秀的Python环境管理软件。MiniConda3最新版只包含conda及其依赖项如果您更愿意拥有conda以及超过720个开源软件包&…
阅读更多...
怎么安装es、kibana（单点安装）

怎么安装es、kibana（单点安装）

1.部署单点es 1.1.创建网络 因为我们还需要部署kibana容器，因此需要让es和kibana容器互联。这里先创建一个网络： docker network create es-net1.2.加载镜像 这里我们采用elasticsearch的7.12.1版本的镜像，这个镜像体积非常大&#xff0c…
阅读更多...
增删改查管理系统 总结1

增删改查管理系统 总结1

//提醒： 管理员也要有增删改查 新增员工代码完善2可能需要用到 目录 细节1 pom文件出现奇怪页面？ 细节2 如何联系DataGrip与idea？ 细节3 Yapi?接口文档？如何有以下画面？ ​细节4 如何将时间转化为好看的时间&…
阅读更多...
PostgreSQL认证考试PGCA、PGCE、PGCM

PostgreSQL认证考试PGCA、PGCE、PGCM

PostgreSQL认证考试PGCA、PGCE、PGCM 【重点！重点！重点！】PGCA、PGCE、PGCM 直通车快速下正，省心省力，每2个月一次考试 PGCE考试通知 （2024） 一、考试概览 （一） 报名要…
阅读更多...
必练的100道C语言程序设计练习题（上）

必练的100道C语言程序设计练习题（上）

前言: 在计算机编程的世界中，C语言一直是一门备受推崇的语言。它的简洁性、高效性以及广泛应用使得学习C语言成为每一位程序员的必由之路。然而，掌握这门语言并不是一蹴而就的事情，它需要不断的练习和实践。为了帮助各位编程爱好者更好地理解…
阅读更多...
kafka之java客户端实战

kafka之java客户端实战

1. kafka的客户端 Kafka提供了两套客户端API，HighLevel API和LowLevel API。 HighLevel API封装了kafka的运行细节，使用起来比较简单，是企业开发过程中最常用的客户端API。 而LowLevel API则需要客户端自己管理Kafka的运行细节，Pa…
阅读更多...
通过shell脚本确定当前平台

通过shell脚本确定当前平台

shell中的变量OSTYPE存储操作系统的名称，也可以使用uname命令来确认当前所在的平台。 shell中的变量HOSTTYPE存储操作系统的架构。 测试代码如下所示： #! /bin/bashecho "use OSTYPE:" if [[ "$OSTYPE" "linux-gnu&quo…
阅读更多...
canvas设置渐变色文字（线性、径向）

canvas设置渐变色文字（线性、径向）

查看专栏目录 canvas示例教程100专栏，提供canvas的基础知识，高级动画，相关应用扩展等信息。canvas作为html的一部分，是图像图标地图可视化的一个重要的基础，学好了canvas，在其他的一些应用上将会起到非常重…
阅读更多...
ssm基于JAVA的酒店客房管理系统论文

ssm基于JAVA的酒店客房管理系统论文

摘 要 现代经济快节奏发展以及不断完善升级的信息化技术，让传统数据信息的管理升级为软件存储，归纳，集中处理数据信息的管理方式。本酒店客房管理系统就是在这样的大环境下诞生，其可以帮助管理者在短时间内处理完毕庞大的数据信息…
阅读更多...
JAVA实现循环日期加一天

JAVA实现循环日期加一天

一、业务背景 现在数据库新增字段需要区分平日(0)和假期(1)的数据，之前有一批去年的数据都没有算过，所以得用日期循环来根据实际的时间来修改对应的数值，废话不多说看具体操作方法。 二、操作方法 // 初始日期 String dateString "20…
阅读更多...
谷粒商城项目|es的应用场景及常见问题

谷粒商城项目|es的应用场景及常见问题

es是什么 es多被用于搜索聚合分析引擎 是分布式的可以高性能查询的引擎 es应用场景 为什么不用MYSQL而用es es将数据存在内存中且可以分布式的存储数据 商品上架 商品在es中的保存 1.在es中建立索引 spu sku spu sku保存在一起防止分布查询 为了防止对象数组扁平化&#xff…
阅读更多...
NetDevOps：华三交换机通过Netmiko或者Nornir获取接口信息通过TextFSM解析报错问题

NetDevOps：华三交换机通过Netmiko或者Nornir获取接口信息通过TextFSM解析报错问题

python代码：实现功能获取交换机接口信息并通过TextFSM进行解析。 from netmiko import Netmiko import textfsm show_intf_cmd_mapping {hp_comware: display interface, }def ssh_device_2_get_intfs(device_type, host, username, password, port):dev_info {d…
阅读更多...
微机原理常考填空总结

微机原理常考填空总结

hello大家好我是吃个西瓜嘤，这篇节只总结微机原理常考填空题都是干货展示常出现的易错点以及微机原理注意事项。 以下仅代表个人发言 #微机原理 正文开始： 1，区分JZ，JNZ技巧 也就是D70用JZ；D71用JNZ。 JZ;条件ZF1时…
阅读更多...
LLM之RAG实战（十四）| 利用LongContextRetriver克服RAG中的中间丢失现象

LLM之RAG实战（十四）| 利用LongContextRetriver克服RAG中的中间丢失现象

人类和大型语言模型（LLM）都有一个共同的行为模式：他们往往擅长处理位于给定内容开头或结尾的信息，而中间的信息往往会被忽视。 来自斯坦福大学、加州大学伯克利分校和Samaya AI的研究人员在论文《Lost in the Middle: How Languag…
阅读更多...
安装MySQL

安装MySQL

采用ubuntu系统，安装MySQL5.7 安装 下载apt仓库文件 #下载apt仓库的安装包，Ubuntu的安装包是.deb文件 wget https://dev.mysql.com/get/mysql-apt-config_0.8.12-1_all.deb配置apt仓库 #使用dpkg命令安装仓库dpkg -i mysql-apt-config_0.8.12-1_all.…
阅读更多...
大模型实战营Day3 作业

大模型实战营Day3 作业

基础作业： 复现课程知识库助手搭建过程 (截图) 进阶作业： 选择一个垂直领域，收集该领域的专业资料构建专业知识库，并搭建专业问答助手，并在 OpenXLab 上成功部署（截图，并提供应用地址&#xf…
阅读更多...
Nacos和Eureka比较、统一配置管理、Nacos热更新、多环境配置共享、Nacos集群搭建步骤

Nacos和Eureka比较、统一配置管理、Nacos热更新、多环境配置共享、Nacos集群搭建步骤

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档 文章目录 前言一、Nacos和eureka的对比二、统一配置管理二、Nacos热更新方式一方式二 三、多环境配置共享四、Nacos集群搭建步骤（黑马springCloud的p29&#xff0…
阅读更多...
最新文章

Copyright @ 2022~2023