kafka基本概念及操作

kafka介绍

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的 （replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、 Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

1.使用场景

  • 日志收集：可以用Kafka收集各种服务的log，通过kafka以统⼀接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。

  • 消息系统：解耦和生产者和消费者、缓存消息等。

  • 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、 搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

  • 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

2.kafka基本概念

image-20230811105243277

整个流程应该是：producer通过网络发送消息到Kafka集群，然后consumer 来进行消费，如下图：

image-20230811105555825

服务端(brokers)和客户端(producer、consumer)之间通信通过TCP协议来完成。

kafka基本使用

1.安装&关闭

以下所有操作全部基于kafka_2.13-3.0.1.tgz （3.0.1版本） 这个版本

配置文件server.properties（主要修改以下配置）

#broker.id属性在kafka集群中必须要是唯⼀
broker.id=0
listeners=PLAINTEXT://xx.xx.xx.xx(服务器内网IP地址):9092
advertised.listeners=PLAINTEXT://xx.xx.xx.xx(服务器对外IP地址):9092
#kafka的消息存储⽂件
log.dir=/usr/local/kafka/data/kafka-logs
#kafka连接zookeeper的地址
zookeeper.connect=192.168.65.60:2181

启动

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

验证

# 查看端口是否占用
netstat -ntlp

或者

进入到zk内查看是否有kafka的节点：/brokers/ids/0

./zkCli.sh

image-20230814134630858

关闭kafka

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

2.创建topic

执行以下命令创建名为“test”的topic，这个topic只有⼀个partition，并且备份因子也设置为 1

./kafka-topics.sh --bootstrap-server kafkahost:9092 --create --topic test --partitions 1 --replication-factor 1-- 新版本的kafka，已经不需要依赖zookeeper来创建topic，新版的kafka创建topic指令如下：
./kafka-topics.sh --bootstrap-server 124.222.253.33:9092 --create --topic test --partitions 1 --replication-factor 1

3.查看kafka中所有的主题

./kafka-topics.sh --bootstrap-server kafkahost:9092 --list./kafka-topics.sh --bootstrap-server 124.222.253.33:9092 --list

4.发送消息

kafka自带了⼀个producer命令客户端，可以从本地文件中读取内容或者以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每⼀个行会被当做成⼀个独立的消息。使用kafka的发送消息的客户端，指定发送到的kafka服务器地址和topic

把消息发送给broker中的某个topic，打开⼀个kafka发送消息的客户端，然后开始用客户端向kafka服务器发送消息

./kafka-console-producer.sh --bootstrap-server 124.222.253.33:9092 --topic test

5.消费消息

消费消息两种方式：

对于consumer，kafka同样也携带了⼀个命令行客户端，会将获取到内容在命令中进行输出，默认是消费最新的消息。 使用kafka的消费者消息的客户端，从指定kafka服务器的指定 topic中消费消息

  1. 从当前主题中的最后⼀条消息的offset（偏移量位置）+1开始消费

    ./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --topic test

  2. 从当前主题中的第⼀条消息开始消费

    ./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --from-beginning --topic test

6.消息的细节

image-20230811094410650

  • 生产者将消息发送给broker，broker会将消息保存在本地的日志文件中
/usr/local/kafka/data/kafka-logs/主题-分区/00000000.log
  • 消息的保存是有序的，通过offset偏移量来描述消息的有序性
  • 消费者消费消息时可以通过offset来描述当前要消费的那条消息的位置

7.单播&多播消息

单播还是多播消息取决于topic有多少消费组

1）单播

如果多个消费者在同⼀个消费组，那么只有⼀个消费者可以收到订阅的topic中的消息。（同⼀个消费组中只能有⼀个消费者收到订阅topic中的消息。）

./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --consumer-property group.id=testGroup --topic test

2）多播

不同的消费组订阅同⼀个topic，那么不同的消费组中只有⼀个消费者能收到消息。实际上也是多个消费组中的多个消费者收到了同⼀个消息。

./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --consumer-property group.id=testGroup1 --topic test
./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --consumer-property group.id=testGroup2 --topic test

3）区别

image-20230811111854111

8.查看消费组详细信息

# 查看当前主题下有哪些消费组
./kafka-consumer-groups.sh --bootstrap-server 124.222.253.33:9092 --list# 查看消费组中的具体信息：⽐如当前偏移量、最后⼀条消息的偏移量、堆积的消息数量
./kafka-consumer-groups.sh --bootstrap-server 124.222.253.33:9092 --describe --group testGroup

image-20230811113410786

  • Currennt-offset：当前消费组的已消费偏移量（最后被消费的消息的偏移量）
  • Log-end-offset：主题对应分区消息的结束偏移量(HW) 【消息总量，最后一条消息偏移量】
  • Lag：当前消费组未消费的消息数（积压消息量）

Kafka中主题和分区的概念

1.主题

主题-topic在kafka中是⼀个逻辑的概念，kafka通过topic将消息进⾏分类。不同的topic会被订阅该topic的消费者消费。

但是有⼀个问题，如果说这个topic中的消息非常非常多，多到需要几T来存，因为消息是会被保存到log日志文件中的。为了解决单个文件过大的问题，kafka提出了Partition分区的概念。

2.分区

1）分区的概念

通过partition将⼀个topic中的消息分区来存储。

这样的好处有多个：

  • 分区存储，可以解决统⼀存储文件过大的问题
  • 提高了读写的吞吐量：读和写可以同时在多个分区中进行

image-20230813215613707

2）创建多分区的主题

./kafka-topics.sh --bootstrap-server 124.222.253.33:9092 --create --topic test1 --partitions 2 --replication-factor 1

3.kafka中消息日志文件中保存的内容

  • 00000.log： 这个文件中保存的就是消息

  • __consumer_offsets-49：

    kafka内部自己创建了__consumer_offsets主题包含了50个分区。这个主题用来存放消费者消费某个主题的偏移量。因为每个消费者都会自己维护着消费的主题的偏移量，也就是说每个消费者会把消费的主题的偏移量自主上报给kafka中的默认主题：consumer_offsets。因此kafka为了提升这个主题的并发性，默认设置了50个分区(可以通过offsets.topic.num.partitions设置)。

    • 提交到哪个分区：通过hash函数：hash(consumerGroupId) % __consumer_offsets 主题的分区数
    • 提交到该主题中的内容是：key是consumerGroupId+topic+分区号，value就是当前offset的值

  • 文件中保存的消息，默认保存7天。七天到后消息会被删除，最后就保留最新的那条数据。

Kafka集群操作

1.搭建kafka集群（三个broker）

  • 创建三个server.properties文件
# 0 1 2
broker.id=2
# 9092 9093 9094
listeners=PLAINTEXT://xx.xx.xx.xx(服务器内网IP地址):9094
advertised.listeners=PLAINTEXT://xx.xx.xx.xx(服务器对外IP地址):9094
# kafka-logs kafka-logs-1 kafka-logs-2
log.dir=/usr/local/data/kafka-logs-2
  • 通过命令来启动三台broker
./kafka-server-start.sh -daemon ../config/server.properties
./kafka-server-start.sh -daemon ../config/server1.properties
./kafka-server-start.sh -daemon ../config/server2.properties
  • 校验是否启动成功

进入到zk中查看/brokers/ids中过是否有三个znode（0，1，2）

2.副本的概念

在创建主题时，除了指明了主题的分区数以外，还指明了副本数 replication-factor参数

如下主题，创建了两分区、三副本（副本对应集群中broker数量）

./kafka-topics.sh --bootstrap-server 124.222.253.33:9092 --create --topic my-replicated-topic --partitions 2 --replication-factor 3

副本是为了为主题中的分区创建多个备份，多个副本在kafka集群的多个broker中，会有⼀个副本作为leader，其他是follower。

查看topic情况：

# 查看topic情况
./kafka-topics.sh --describe --bootstrap-server 124.222.253.33:9092 --topic my-replicated-topic

image-20230814105758165

image-20230814105921066

  • leader：

kafka的写和读的操作，都发生在leader上。leader负责把数据同步给follower。当leader挂了，经过主从选举，从多个follower中选举产⽣⼀个新的leader（follower通过poll的方式来同步数据）

  • follower：

接收leader的同步的数据，leader挂了，参与leader选举

  • replicas：

当前副本存在的broker节点

  • isr：

可以同步和已同步的broker节点会被存入到isr集合中。如果isr中的broker节点性能较差，会被踢出isr集合。

3.broker、主题、分区、副本

综上broker、主题、分区、副本概念已全部展示：

集群中有多个broker，创建主题时可以指明主题有多个分区（把消息拆分到不同的分区中存储），可以为分区创建多个副本，不同的副本存放在不同的broker⾥。

4.kafka集群消息的发送

./kafka-console-producer.sh --broker-list 124.222.253.33:9092,124.222.253.33:9093,124.222.253.33:9094 --topic my-replicated-topic

5.kafka集群消息的消费

1）普通消费

./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092,124.222.253.33:9093,124.222.253.33:9094 --from-beginning --topic my-replicated-topic

2）指定消费组消费

./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092,124.222.253.33:9093,124.222.253.33:9094 --from-beginning --consumer-property group.id=testGroup1 --topic my-replicated-topic

6.分区分消费组的集群消费中的细节

image-20230814115045080

  • ⼀个partition只能被⼀个消费组中的⼀个消费者消费，目的是为了保证消费的顺序性，但是多个partion的多个消费者消费的总的顺序性是得不到保证的，那怎么做到消费的总顺序性呢？（Kafka只在partition的范围内保证消息消费的局部顺序性，不能在同⼀个topic中的多个partition中保证总的消费顺序性。 ⼀个消费者可以消费多个partition。）

  • partition的数量决定了消费组中消费者的数量，建议同⼀个消费组中消费者的数量不要超过partition的数量，否则多的消费者消费不到消息

  • 如果消费者挂了，那么会触发rebalance机制，会让其他消费者来消费该分区

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

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

相关文章

java毕业设计-智慧食堂管理系统-内容快览

java毕业设计-智慧食堂管理系统-内容快览

首页 智慧食堂管理系统是一种可以提高食堂运营效率的管理系统。它将前端代码使用Vue实现，后端使用Spring Boot实现。这个系统的目的是简化食堂管理，提高食堂服务质量。在现代快节奏的生活中，人们对餐饮服务提出了更高的要求，食堂管…
阅读更多...
457. 环形数组是否存在循环

457. 环形数组是否存在循环

457. 环形数组是否存在循环 原题链接：完成情况：解题思路：参考代码：经验吸取 原题链接： 457. 环形数组是否存在循环 https://leetcode.cn/problems/circular-array-loop/description/ 完成情况： 解题思路…
阅读更多...
undefined reference to `dlopen‘ ‘SSL_library_init‘ `X509_certificate_type‘

undefined reference to `dlopen‘ ‘SSL_library_init‘ `X509_certificate_type‘

使用Crow的时候需要注意crow依赖asio依赖OpenSSL，asio要求1.22以上版本，我使用的是1.26.0； 这个版本的asio要求OpenSSL是1.0.2，其他版本我得机器上编不过，ubuntu上默认带的OpenSSL是1.1.1; 所以我下载了OPENSSL1.2.0重…
阅读更多...
图像变形之移动最小二乘算法（MLS）

图像变形之移动最小二乘算法（MLS）

基本原理 基于移动最小二乘的图像变形是通过一组源控制点和目标控制点来控制变形，对于每一个待求变形后位置的点而言，根据预设的形变类型（如仿射变换、相似变换、刚性变换）求解一个最小二乘优化目标函数估计一个局部的坐标变换矩阵…
阅读更多...
Android自定义侧滑Item

Android自定义侧滑Item

源码地址&#xff1a;https://github.com/LanSeLianMa/CustomizeView/tree/master/cehuaitem 使用方式一&#xff1a;XML布局中直接使用 <?xml version"1.0" encoding"utf-8"?> <LinearLayout xmlns:android"http://schemas.android.com…
阅读更多...
linux系统服务学习（一）Linux高级命令扩展

linux系统服务学习（一）Linux高级命令扩展

文章目录 Linux高级命令&#xff08;扩展&#xff09;一、find命令1、find命令作用2、基本语法3、*星号通配符4、根据文件修改时间搜索文件☆ 聊一下Windows中的文件时间概念&#xff1f;☆ 使用stat命令获取文件的最后修改时间☆ 创建文件时设置修改时间以及修改文件的修改时间…
阅读更多...
服务器数据恢复-RAID5多块磁盘离线导致崩溃的数据恢复案例

服务器数据恢复-RAID5多块磁盘离线导致崩溃的数据恢复案例

服务器数据恢复环境&#xff1a; DELL POWEREDGE某型号服务器中有一组由6块SCSI硬盘组建的RAID5阵列&#xff0c;LINUX REDHAT操作系统&#xff0c;EXT3文件系统&#xff0c;存放图片文件。 服务器故障&分析&#xff1a; 服务器raid5阵列中有一块硬盘离线&#xff0c;管理员…
阅读更多...
ChatGPT收录

ChatGPT收录

VSCode插件-ChatGPT 多磨助手 多磨助手 (domore.run) Steamship Steamship 免费合集 免费chatGPT - Ant Design Pro 免费AI聊天室 (xyys.one)
阅读更多...
SpringBoot复习：（33）WebMvcAutoconfiguration内部静态类WebMvcAutoConfigurationAdapter

SpringBoot复习：（33）WebMvcAutoconfiguration内部静态类WebMvcAutoConfigurationAdapter

WebMvcAutoconfiguration内部静态类WebMvcAutoConfigurationAdapter实现了WebMvcConfigurer接口&#xff0c;重写了一些方法&#xff0c;也就是默认对Spring Mvc进行了一些配置: 该静态类上有个**Import**注解&#xff1a; Import(EnableWebMvcConfiguration.class) 它的父类…
阅读更多...
Java课题笔记~ 过滤器

Java课题笔记~ 过滤器

概念 过滤器&#xff1a;顾名思义&#xff0c;就是在源数据和目的数据之间起过滤作用的中间组件。 过滤器可以动态地拦截请求和响应&#xff0c;以变换或使用包含在请求或响应中的信息。 filter是对客户端访问资源的过滤&#xff0c;符合条件放行&#xff0c;不符合条件不放…
阅读更多...
MFC编程基础

MFC编程基础

一、MFC类库简介 ●微软基础类库( Microsoft Foundation Classes, MFC )是Microsoft公司提供的关于Win32 API的C类库集&#xff0c;它定义了一个标准的应用程序框架。 ●借助这个框架,用户可以轻松地开发标准的Windows应用程序。 ●MFC实现了标准的用户…
阅读更多...
SAP ABAP 直接把内表转换成PDF格式（smartform的打印函数输出OTF格式数据）

SAP ABAP 直接把内表转换成PDF格式（smartform的打印函数输出OTF格式数据）

直接上代码&#xff1a; REPORT zcycle055.DATA: lt_tab TYPE TABLE OF zpps001. DATA: ls_tab TYPE zpps001.ls_tab-werks 1001. ls_tab-gamng 150.00. ls_tab-gstrp 20201202. ls_tab-aufnr 000010000246. ls_tab-auart 标准生产. ls_tab-gltrp 20201205. ls_tab-matn…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023