0


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--topictest--partitions1 --replication-factor 1

-- 新版本的kafka,已经不需要依赖zookeeper来创建topic,新版的kafka创建topic指令如下:
./kafka-topics.sh --bootstrap-server 124.222.253.33:9092 --create--topictest--partitions1 --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 --topictest

5.消费消息

消费消息两种方式

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

  1. 从当前主题中的最后⼀条消息的offset(偏移量位置)+1开始消费./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --topictest
  2. 从当前主题中的第⼀条消息开始消费./kafka-console-consumer.sh --bootstrap-server 124.222.253.33:9092 --from-beginning --topictest

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 --topictest

2)多播

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

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

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 --partitions2 --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 --partitions2 --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机制,会让其他消费者来消费该分区

kafka集群中的controller、 rebalance、HW

1.controller

  • 集群中谁来充当controller 【Kafka集群中始终只有一个Controller Broker

每个broker启动时会向zk创建⼀个临时序号节点,获得的序号最小的那个broker将会作为集群中的controller

负责这么几件事:

  • 当集群中有⼀个副本的leader挂掉,需要在集群中选举出⼀个新的leader,选举的规则是 从isr集合中最左边获得。
  • 当集群中有broker新增或减少,controller会同步信息给其他broker
  • 当集群中有分区新增或减少,controller会同步信息给其他broker
脑裂

如果controller Broker 挂掉了,Kafka集群必须找到可以替代的controller,集群将不能正常运转。这里面存在一个问题,很难确定Broker是挂掉了,还是仅仅只是短暂性的故障。但是,集群为了正常运转,必须选出新的controller。如果之前被取代的controller又正常了,他并不知道自己已经被取代了,那么此时集群中会出现两台controller
其实这种情况是很容易发生。比如,某个controller由于GC而被认为已经挂掉,并选择了一个新的controller。在GC的情况下,在最初的controller眼中,并没有改变任何东西,该Broker甚至不知道它已经暂停了。因此,它将继续充当当前controller,这是分布式系统中的常见情况,称为脑裂。

2.rebalance机制

  • 前提:消费组中的消费者没有指明分区来消费
  • 触发的条件:当消费组中的消费者和分区的关系发生变化的时候
  • 分区分配的策略:在rebalance之前,分区怎么分配会有这么三种策略 - range:根据公式计算得到每个消费消费哪几个分区:前面的消费者是分区总数/消费者数量+1,之后的消费者是分区总数/消费者数量(7个分区,三个消费者,则是:第一个消费者前三个分区,后面两个消费者各两个分区;8,3->3,3,2;9,3->3,3,3)- 轮询:大家轮着来- sticky:粘合策略,如果需要rebalance,会在之前已分配的基础上调整,不会改变之前的分配情况。如果这个策略没有开,那么就要进行全部的重新分配(全部重新分配效率较差)。建议开启。

3.HW和LEO

LEO是某个副本最后消息的消息位置(log-end-offset)

HW是已完成同步的位置。消息在写入broker时,且每个broker完成这条消息的同步后,hw才会变化。在这之前消费者是消费不到这条消息的。在同步完成之后,HW更新之后,消费者才能消费到这条消息,这样的目的是防止消息的丢失。(保证了如果leader所在的broker失效,该消息仍然可以从新选举的leader中获取。)

HW <= LEO

image-20230815233558072

标签: kafka 分布式

本文转载自: https://blog.csdn.net/qq_43417581/article/details/132274507
版权归原作者 Likelong~ 所有, 如有侵权,请联系我们删除。

“kafka基本概念及操作”的评论:

还没有评论