Kafka的简介及架构

目录

消息队列

产生背景

消息队列介绍

常见的消息队列产品

应用场景

 消息队列的消息模型

Kafka的基本介绍

简介

Kafka的架构

Kafka的使用

Kafka的shell命令

Kafka的Python API的操作

完成生产者代码

完成消费者代码

消息队列

产生背景

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

消息:指的数据,只不过这个这个数据存在一定流动状态

队列:指的容器,可以存储数据,这个容器具备FIFO(先进先出)特性

公共容器的特点:

1.公共性:各个程序都可以与之对接

2.FIFO特性:先进先出

3.具备高效的并发能力:能够承载海量数据

4.具备一定的容错能力:比如支持重新读取消息方案

消息队列介绍

常见的消息队列产品

MQ：message queue消息队列

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

应用场景

消息队列的应用场景:

1.应用解耦合

2.异步处理

3.限流削峰

4.消息驱动系统

 消息队列的消息模型

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

消息队列中两个角色:生产者(producer)和消费者(consumer)

生产者:生产/发送消息到消息队列中

消费者:从消息队列中获取消息

在JMS规范中,专门规定了两种消息消费类型:

1.点对点消费类型:一条消息最终只能被一个消费所消费,微信聊天的私聊

2.发布订阅消费模型:指一条消息最终被多个消费者所消费,微信聊天的群聊

Kafka的基本介绍

简介

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

Kafka的特点:

1.可靠性:Kafka集群是分布式的,有多副本机制,数据可以自动复制

2.可扩展性:Kafka集群可以灵活的调整,在线扩容

3.耐用性:Kafka数据保存在磁盘上,数据有多副本机制,数据持久化,一定程度上防止数据丢失

4.高性能:Kafka可以存储海量的数据,虽然是使用磁盘进行存储,但是Kafka有各种优化手段(例如：磁盘的顺序读写,零拷贝等)提高数据的读写速度(吞吐量)

Kafka的架构

1. Kafka中集群节点叫broker,节点与节点之间没有主从之分,地位是完全一样

2.Topic:主题/话题,是业务层面对消息进行分类的

3.一个Topic可以设置多个分区

4.同一个partition分区可以设置多个副本,但是副本数不能超过(>)集群broker节点的个数

5.broker节点间没有主从之分,但是同一个partition分区的不同副本间有主从之分,分为Leader主副本和Follwer从副本

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本质上是一个消息队列中间件产品,主要负责消息数据的传递,也就说学习Kafka 也就是学习如何使用Kafka生产数据，以及如何使用Kafka来消费数据

创建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的分区数

        --replicattion-factor:设置Topic分区的的副本数

注意:如果副本数超过了集群broker节点个数,会报错

查看Topic

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

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

查看具体Topic

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

模拟生产者Producer

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

 模拟消费者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的数据量是特别大的，消费、打印都需要消耗服务器的资源，如果不限定消费的最大条数，可能造成服务器宕机。

修改Topic

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

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

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

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

Kafka的Python API的操作

准备工作：在服务器的节点上安装一个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 timefrom 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.topicpartition = metadata.partition"""offset消息偏移量，从0开始编号。也就是一条消息在分区中的序号/索引在不同分区间，消息偏移量是无序在同一个分区里面，消息偏移量是有序"""offset = metadata.offsetprint(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()

完成消费者代码

from kafka import KafkaConsumerif __name__ == '__main__':# 1- 创建消费者consumer = KafkaConsumer("test01",bootstrap_servers=["node1.itcast.cn:9092", "node2.itcast.cn:9092"])# 2- 消费消息for msg in consumer:topic = msg.topicpartition = msg.partitionoffset = msg.offset# key和value消费出来都是bytes数据类型，需要进行解码key = msg.keyvalue = msg.valueprint(f"{topic}，{partition}，{offset}，{key}，{value.decode('UTF-8')}，{msg}")

 

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

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

相关文章

PyCharm使用手册

PyCharm使用手册

配置文件和代码模板 文件注释模板： 注释项描述示例Project项目名称hello_pythonFile文件名称hello_python.pyAuthor作者Zhang SanDate创建时间2024-01-11 17:05:00PyVersionPython解释器版本Python3.7Description文件描述这是一个python语言入门文件 效果示例&am…
阅读更多...
【阅读笔记】Chain of LoRA

【阅读笔记】Chain of LoRA

一、论文信息 1 论文标题 Chain of LoRA: Efficient Fine-tuning of Language Models via Residual Learning 2 发表刊物 arXiv2023 3 作者团队 Department of Computer Science, Princeton University School of Computer Science and Engineering, Nanyang Technologic…
阅读更多...
2023年人工智能的最新发展（下）

2023年人工智能的最新发展（下）

目录 1.MidJourney： 2.GAN: 3.Diffusion Model 4.DALLE、Disco Diffusion 5.Stable Diffusion 1.MidJourney： 2023年3月，一组中国小情侣的照片在网络上迅速走红。这组照片看起来普通，就像一对小情侣的合影，但实…
阅读更多...
Go后端开发 -- 条件、循环语句 defer语句

Go后端开发 -- 条件、循环语句 defer语句

Go后端开发 – 条件、循环语句 && defer语句 文章目录 Go后端开发 -- 条件、循环语句 && defer语句一、条件语句1.if ... else 语句2.switch语句3.select语句 二、循环语句1.for循环 三、defer语句1.defer语句的作用2.defer和return的先后顺序3.recover错误拦截…
阅读更多...
212. 单词搜索 II（字典树的另一种类型）

212. 单词搜索 II（字典树的另一种类型）

大致思路是： 根据words列表建立字典树，其中注意在单词末尾，将原来的isEnd变量换成存储这个单词的变量，方便存储到ans中，另外，字典树的字节点由原来的Trie数组变为hashmap，方便检索字母。 建立…
阅读更多...
C++ n皇后问题 || 深度优先搜索模版题

C++ n皇后问题 || 深度优先搜索模版题

n− 皇后问题是指将 n 个皇后放在 nn 的国际象棋棋盘上，使得皇后不能相互攻击到，即任意两个皇后都不能处于同一行、同一列或同一斜线上。 现在给定整数 n ，请你输出所有的满足条件的棋子摆法。 输入格式 共一行，包含整数 n 。 …
阅读更多...
动态规划day03

动态规划day03

343. 整数拆分(第二次做还是没弄明白) 力扣题目链接(opens new window) 给定一个正整数 n，将其拆分为至少两个正整数的和，并使这些整数的乘积最大化。 返回你可以获得的最大乘积。 示例 1: 输入: 2输出: 1解释: 2 1 1, 1 1 1。 示例 2: 输入: …
阅读更多...
黑马程序员JavaWeb开发|案例：tlias智能学习辅助系统（4）员工管理|修改员工、配置文件

黑马程序员JavaWeb开发|案例：tlias智能学习辅助系统（4）员工管理|修改员工、配置文件

指路（1）（2）（3）👇 黑马程序员JavaWeb开发|案例：tlias智能学习辅助系统（1）准备工作、部门管理_tlias智能学习辅助系统的需求分析-CSDN博客https://blog.csdn.n…
阅读更多...
AI人工智能的发展趋势及未来展望

AI人工智能的发展趋势及未来展望

人工智能（Artificial Intelligence，简称AI）作为一门拥有悠久历史但最近才得到广泛关注的领域，正在以惊人的速度推动着科技进步和社会变革。近年来，随着计算能力、数据规模和算法研究的不断突破，人工智能已经…
阅读更多...
java通过HttpClient方式实现https请求的工具类（绕过证书验证）

java通过HttpClient方式实现https请求的工具类（绕过证书验证）

目录 一、引入依赖包二、HttpClient方式实现的https请求工具类三、测试类 一、引入依赖包 引入相关依赖包 <!--lombok用于简化实体类开发--><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><option…
阅读更多...
MySQL批量插入技巧

MySQL批量插入技巧

关于MySQL批量插入的一些问题 MySQL一直是我们互联网行业比较常用的数据&#xff0c;当我们使用半ORM框架进行MySQL大批量插入操作时&#xff0c;你是否考虑过这些问题: 进行大数据量插入时&#xff0c;是否需要进行分批次插入&#xff0c;一次插入多少合适&#xff1f;有什么…
阅读更多...
泡泡玛特台北旗舰店打造“乐园式体验”，打造西门町新地标

泡泡玛特台北旗舰店打造“乐园式体验”，打造西门町新地标

近日&#xff0c;泡泡玛特台北旗舰店盛大开业&#xff0c;矗立于西门町核心商圈的壮观五层独栋建筑吸引大量游客驻足。作为年度收官之作&#xff0c;该店总面积700多平方米&#xff0c;售卖超过千款潮玩。各楼层经过精心规划&#xff0c;除了各大IP最新款潮玩产品及衍生品之外&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023