当前位置:   article > 正文

[docker-compose中间件集群系列] - 4 kafka的作用、原理以及集群搭建_kafka 的集群作用

kafka 的集群作用

上一篇介绍了zookeeper的作用以及原理,这一篇我们介绍消息中间件kafka以及为什么要用kafka,什么业务场景下需要kafka,还有不同业务场景应该用哪一种消息中间件。最后还有kafka集群的搭建。注,kafka从2.8版本开始已经在逐步取消依赖zookeeper了,本文依赖zookeeper以及不依赖zookeeper改用Kraft模式都会介绍。


kafka是什么

在说kafka是什么之前,我们应该先知道,消息中间件是什么。举个例子,大家网上买东西,现在很多时候都会放到驿站或者快递柜,而不是快递员直接送到你手上,然后你什么时候去取快递,那就是你的事了,那在这个过程中,中间的这个驿站或者快递柜,就是充当了快递中间件的角色,快递员就是生产者,而你,就是消费者了。那消息中间件,在我们系统架构的层面上去理解,就是消息先到消息中间件,然后消费者服务再去消息中间件消费消息这样的一个过程。而kafka是众多消息中间件里的其中一种。


kafka能干什么

kafka作为一个发布订阅消息系统,它有几个特性,分别是消息持久化,高吞吐,分布式,多客户端支持,实时等,适用于离线和在线的消息消费。kafka能支持百万级的消息写入和读取吞吐量,并且能在分布式环境中维护消息的顺序性和一致性。还支持数据备份和容错,可以保证数据的可靠性。在我们的系统有很多业务场景都可以用到kafka,例如:

  • 日志收集,各种日志都可以发送到kafka,以topic分类然后再给各种consumer消费分析。
  • 消息系统,解耦生产者和消费者,缓存消息。
  • 削峰,减低下游消息处理的压力。
  • 流式处理,spark等。
  • 事件源。

kafka的原理

在了解kafka原理之前,我们先把涉及到的相关基本概念先弄清楚。

  • broker:kafka实例。假设我们有3台服务器,每一台服务器启动一个kafka实例,那每一台服务器相当于1个broker。几个broker共同构成我们的kafka集群。
  • controller:在kafka集群中,某一个broker将被选举出来担任一个特殊角色,用于管理和协调kafka集群,管理集群中所有分区的状态并执行相应的管理操作。每一个kafka集群任意时刻都只能有一个controller。当集群启动的时候,所有broker都参与controller的竞选,最终有一个胜出,一旦controller在某个时刻崩溃,集群中的其他broker会收到通知,然后开始新的一轮controller竞选,新的controller将承担之前controller的所有工作。controler的详细作用我们后面说。
  • topic:主题。在kafka中,使用一个类别属性来划分消息的所属类别,这个类别称为topic。
  • partition:分区。topic中的消息会被分割为一个或多个的分区,它是一个物理概念,对应到系统上就是一个或若干个目录,一个分区就是一个文件目录。消息以追加的形式写入分区。分区的目录命名一般以topic+partition的形式。由于一个topic可以包含多个分区,因此无法保证消息在整个topic中有序,但是单个分区内可以保证有序,单个topic的不同分区可以分布在不同的broker上,也就是说,一个topic可以跨越多个服务器,但同一个分区,不可能存在于多个broker中。kafka支持动态添加partition,但不支持删减partition,因为会影响其他partition中消息的有序性。
  • segment:段。物理概念,一个分区(即目录)中的数据会被划分到不同的文件当中,所有分区中segment的文件都是用20位编号来命名文件的。当文件达到一定条件后不再写入,开始创建另外的segment文件进行存储。每一段segment包含3种文件,*.log、*.index/.timeindex。3种文件共同被称为一个segment(1个index文件+多个log文件)。其中*.log存储真正的数据,*.index/.timeindex存储对应*.log文件的索引。index文件中存放的是offset+偏移量,log文件中存放的是偏移量和message的映射关系,文件的命名以当前文件存储的最小offset命名。假设我们要实现从指定offset开始读取数据,只需要根据给定的offset与对应partition下的segment文件名做对比,然后根据*.index文件查找给定offset的值所对应的实际磁盘偏移量,再拿着偏移量去*.log文件读取目标数据即可,查询顺序是offset->偏移量->message,其中offset->偏移量使用二分查找,偏移量->message使用线性查找
  • offset:每一条消息都有的一个唯一编号。而每一个消费者又维护着一个消费序号。消费者维护的消费序号,我们又分为current offset和commit offset。消费者自己维护的序号,和消息实际的序号是两种东西,请不要搞混。
  • replication:副本集。replication逻辑上是作用于topic的,但是实际上是体现在每一个partition中的。每一个partition都会有多个replication,但其中,会区分出1个leader和其余follower,producer和consumer只会和分区当中的leader交互,而follower则只从leader中同步数据,同步会存在一定的滞后性。所以如果看见什么leader partition,其实和leader replication是同一回事,因为副本集策略只是一种机制,主要是为了提高可用性。
  • producer:消息的产生者
  • consumer:消息的消费者
  • consumer group:消费组。一个消费组当中,可以有多个消费者,消费者可以是进程,也可以是线程,一个消费组,用group id作为消费组的唯一标识,消费组下订阅的topic的每一个分区,只能分配给这一个消费组中的某一个消费者消费。当然,其他消费组也可以同时订阅同一个topic。不同消费组消费的情况互相隔离。当然,其中涉及到的分区和消费者数量的不一致的几种情况,我们放后面说。
  • message:消息。消息发布到topic,会将key hash一下,得到分区编号,决定放到哪一个分区,如果没写key,则会轮询出一个分区。无论消息是否被消费,kafka都会保存所有消息,对于旧数据有两种处理方式,一种是删除,另一种是压缩。对于删除策略也有两种处理方式,一种是基于时间,默认是168小时即7天,一种是基于大小,默认是1073741824byte。另外,压缩策略则是只保留每一个key的最后一个版本的数据。

基本概念差不多就这些,下面我们用图来表示一下他们之间的关系。

先宏观的说一下producer、kafka cluster、consumer三者的关系以及数据流的方向。

如上图表示:

  1. producer、kafka cluster、consumer之间的关系;
  2. kafka cluster的构成(多个broker共同组成cluster);
  3. consumer group的构成(多个consumer实例共同构成消费组);
  4. topic内partition的分布(分区有可能存在不同broker中,也有可能在同一broker中,但同一分区不可能在不同broker中);
  5. producer1向topic1推送消息,以及consumer group2消费topic3的数据流向。(消费者小于分区时,一个消费者会消费多个分区)

然后说一下topic、partition、replication(leader)、replication(follower)之间的关系。

如上图表示:

  1. 一个topic可以有多个partition
  2. 不同partition可以在不同broker上,也可以在同一个broker上
  3. 每一个partition可能有多个replication,但leader只能有1个,其余为follower。follower同步leader的数据。consumer在旧版本只能从leader读取数据,但新版本也可从follower读取数据。(实现了找到"最近"的replication算法)。实际上与partition交互,也可以看作是和leader replication交互。

接下来说一下broker是如何选出controller的,以及选出来的controller有什么作用。

controller可以做replication状态管理

        维护的状态分为两类:

                1.每台broker上的分区replication信息

                2.每个分区的leader replication信息

        kafka为replication定义了7种状态:

                1.new replication,创建副本时的最初状态,当处于这个状态,replication只能成为follower replication。

                2.online replication,replication启动后,就会变成该状态,处于这个状态,replication可变为follower或leader。

                3.offline replication,replication所在的broker崩溃,会变为这个状态。

                4.replication delete started,开始topic得删除操作后,topic下得所有分区replication都会变为这个状态。

                5.replication deletion successful,若replication成功响应了删除replication请求,进入该状态。

                6.replication deletion ineligible,replication删除失败,进入该状态。

                7.non existent replication,replication删除成功,进入该状态。

        replication状态流转过程:

                1.topic新建时,该topic下得所有replication都是non existent replication状态。

                2.controller加载zookeeper中该topic每个partition的所有replication信息到内存中,同时将该replication状态变为new replication。随后controller将该partition中的第一个replication作为leader然后所有replication被设置进入ISR(同步副本集),然后在zookeeper持久化这个信息。

                3.当确定了partition的leader和ISR后,controller将这些信息以请求的方式发送给所有replication,同时将replication状态同步到集群的所有broker。最后controller将partition的所有replication状态设置为online。

                4.当开始topic删除时,controller尝试停止所有replication,此时replication会停止向leader获取数据。如果停止的是leader本身,则controller会设置该partition的leader为no leader,之后replication进入offline状态。当所有replication都offline时,controller将replication状态变为deletion started表示topic删除任务开始,在这一步的过程中,controller向这些partition得broker发送请求,让他们删除本机上的replication数据。删除成功后,状态变为deletion successful。失败的会进入deletion ineligible,等待controller重试。后续会变为non existent状态,同时controller的上下文缓存会清除这个replication信息。

controller可以做partition状态管理

        partiiton状态如下:

                1.non existent,表示不存在或者已删除的分区。

                2.new partition,partition被创建时,处于这种状态,此时kafka为该分区确认partition列表,但还没选出leader partition和ISR(同步副本集)。然后在zookeeper持久化这个信息。

                3.online partition,一旦确认了leader,进入该状态,正常工作的partition都在这一个状态。

                4.offline partition,在正常工作的过程中,所在broker宕机,则partition进入此状态。

        partition状态流转过程:

                1.新建一个topic时,controller负责创建partition对象,并设置partition为non existent,随后controller的上下文信息读取replication分配方案,最后将partition状态设置为new。

                2.处于new状态的partition还没有leader和ISR,因此controller会初始化leader和ISR并设置partition状态为online,此时partition开始正常工作。

                3.若用户发起topic删除或者关闭broker操作,controller会将收影响的partition设置为offline状态。如果是删除topic,controller会开启partition下所有replication的删除操作,并最终将partition设置为non existent。

所以controller的职责有几个

        1.更新集群元数据信息        2.创建topic        3.删除topic        4.partition重分配        5.leader选举        6.topic扩展         7.broker加入集群        8.broker崩溃处理        9.受控关闭        10.controller leader选举

然后说一下partition和consumer之间的数量关系分别会导致什么结果。

1.consumer大于partition

同一个partition只能被同一个consumer group内的某一个consumer消费,多出来的consumer空闲。会造成资源浪费。

2.consumer小于等于partition

一个consumer对应多个partition,consumer2接收到的消息量是consumer1的两倍,会导致consumer group消费压力不均。多个consumer group相互隔离。

因此,consumer和partition一一对应的状态,是比较推荐的。


 kafka为什么会依赖zookeeper以及为什么要弃用zookeeper

        为什么要用zookeeper上面已经说过了,需要同步集群信息。那为什么要弃用zookeeper的原因也比较简单,一个中间件需要依赖另一个中间件。首先就比较奇怪。然后还有zookeeper有个特点,强一致性。如果zookeeper集群中的某个节点数据发生更变,那么会通知其他节点同时执行更新,那需要等超过半数都更新成功,那这写入性能就较差了。另外,zookeeper只适用于存储一些简单的配置或者集群的元数据,不是真正意义上的存储系统。如果写入数据量过大,zookeeper得性能和稳定性就会下降,可能导致watch延迟或者丢失。所以在kafka集群较大,partition也很多的时候,zookeeper的元数据就会很多,性能就很差,还有zookeeper也是分布式,也是走选举这种逻辑的,他的选举也不是特别快,而且发生选举的时候,整个集群是不对外提供服务的。

        因此,kafka没有zookeeper之后,就把元数据存储到自己内部,利用之前的log存储机制来保存元数据。会有一个元数据topic,元数据会像普通message一样保存在log里面,另外还搞了个KRaft来实现controller quorum。

        依赖zookeeper和不依赖zookeeper在下面的搭建过程中均有涉及。


kafka集群搭建

这次我们还是用server02、server03、server04来搭建,因为上一篇已经搭建了zookeeper集群了,我们先搭建有zookeeper的版本,然后再搭建没有zookeeper的版本。其实也就是docker-compose配置的不同而已,因为是用docker-compose,down掉也比较容易。

第一步先去dns服务器配好我们的kafka域名解析,

kafka01.xdeas.com.cn、zookeeper01.com.cn

kafka02.xdeas.com.cn、zookeeper02.com.cn

kafka03.xdeas.com.cn、zookeeper02.com.cn

分别对应我们的server02、server03、server04。

  

zookeeper版本:

我们还是先创建好需要的文件夹。

然后写我们的docker-compose.yaml文件 

配置文件中有不一样的地方如图所示。然后跑一下 docker-compose up -d 就可以了。

这里贴出三份配置文件

server02

  1. version: '3'
  2. services:
  3. kafka-cluster:
  4. image: bitnami/kafka:3.2.0
  5. restart: always
  6. container_name: kafka-cluster
  7. user: "1001:0"
  8. ports:
  9. - 9092:9092
  10. environment:
  11. - KAFKA_CFG_BROKER_ID=1
  12. - KAFKA_HEAP_OPTS=-Xmx1g -Xms1g
  13. - ALLOW_PLAINTEXT_LISTENER=yes
  14. - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
  15. - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka01.com.cn:9092
  16. - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper01.com.cn:2181,zookeeper02.com.cn:2181,zookeeper03.com.cn:2181
  17. - KAFKA_CFG_NUM_PARTITIONS=10
  18. - KAFKA_CFG_NUM_IO_THREADS=8
  19. - KAFKA_CFG_NUM_NETWORK_THREADS=3
  20. volumes:
  21. - "/etc/localtime:/etc/localtime"
  22. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/data:/bitnami/kafka/data'
  23. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/logs:/bitnami/kafka/logs'

server03

  1. version: '3'
  2. services:
  3. kafka-cluster:
  4. image: bitnami/kafka:3.2.0
  5. restart: always
  6. container_name: kafka-cluster
  7. user: "1001:0"
  8. ports:
  9. - 9092:9092
  10. environment:
  11. - KAFKA_CFG_BROKER_ID=2
  12. - KAFKA_HEAP_OPTS=-Xmx1g -Xms1g
  13. - ALLOW_PLAINTEXT_LISTENER=yes
  14. - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
  15. - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka02.com.cn:9092
  16. - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper01.com.cn:2181,zookeeper02.com.cn:2181,zookeeper03.com.cn:2181
  17. - KAFKA_CFG_NUM_PARTITIONS=10
  18. - KAFKA_CFG_NUM_IO_THREADS=8
  19. - KAFKA_CFG_NUM_NETWORK_THREADS=3
  20. volumes:
  21. - "/etc/localtime:/etc/localtime"
  22. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/data:/bitnami/kafka/data'
  23. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/logs:/bitnami/kafka/logs'

server04

  1. version: '3'
  2. services:
  3. kafka-cluster:
  4. image: bitnami/kafka:3.2.0
  5. restart: always
  6. container_name: kafka-cluster
  7. user: "1001:0"
  8. ports:
  9. - 9092:9092
  10. environment:
  11. - KAFKA_CFG_BROKER_ID=3
  12. - KAFKA_HEAP_OPTS=-Xmx1g -Xms1g
  13. - ALLOW_PLAINTEXT_LISTENER=yes
  14. - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
  15. - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka03.com.cn:9092
  16. - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper01.com.cn:2181,zookeeper02.com.cn:2181,zookeeper03.com.cn:2181
  17. - KAFKA_CFG_NUM_PARTITIONS=10
  18. - KAFKA_CFG_NUM_IO_THREADS=8
  19. - KAFKA_CFG_NUM_NETWORK_THREADS=3
  20. volumes:
  21. - "/etc/localtime:/etc/localtime"
  22. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/data:/bitnami/kafka/data'
  23. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/logs:/bitnami/kafka/logs'

搭建成功之后,我们上去zookeeper查看broker信息。

通过 docker exec -ti zookeeper /bin/bash zkCli.sh -server 进入zk客户端

然后 ls /brokers/ids 看见如下图[1,2,3]就行了,注意这个[/brokers/ids]路径,如果你kafka启动的docker-compose文件里面zookeeper路径是zookeeper.com.cn:2181/kafka,你ls的路径就是/kafka/brokers/ids。

Kraft版本:

接下来我们搭建不依赖zookeeper的版本,由于我懒得搭两个kafka了,所以我们先docker-compose down掉之前的kafka。然后进入volumes文件夹把data文件夹里面的内容删掉。然后把上一份docker-compose.yaml文件改名改成zookeeper_version.yaml备份一下然后cp一份docker-compose出来就行了。我们直接改docker-compose.yaml文件的内容。

Kraft版本的docker-compose内容,最后的那个cluster id,必须一致。必填

改动已经在图中用箭头以及红框指出。这里贴出三份配置文件。

server02

  1. version: '3'
  2. services:
  3. kafka-cluster:
  4. image: bitnami/kafka:3.2.0
  5. restart: always
  6. container_name: kafka-cluster
  7. user: "1001:0"
  8. ports:
  9. - 9092:9092
  10. - 9093:9093
  11. environment:
  12. - KAFKA_CFG_BROKER_ID=1
  13. - KAFKA_HEAP_OPTS=-Xmx1g -Xms1g
  14. - ALLOW_PLAINTEXT_LISTENER=yes
  15. - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
  16. - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
  17. - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka01.com.cn:9092
  18. - KAFKA_CFG_NUM_PARTITIONS=10
  19. - KAFKA_CFG_NUM_IO_THREADS=8
  20. - KAFKA_CFG_NUM_NETWORK_THREADS=3
  21. - KAFKA_ENABLE_KRAFT=yes
  22. - KAFKA_CFG_PROCESS_ROLES=broker,controller
  23. - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
  24. - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka01.com.cn:9093,2@kafka02.com.cn:9093,3@kafka03.com.cn:9093
  25. - KAFKA_KRAFT_CLUSTER_ID=1TGYcbFuRXa4Lqojs4B9Hw
  26. volumes:
  27. - "/etc/localtime:/etc/localtime"
  28. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/data:/bitnami/kafka/data'
  29. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/logs:/bitnami/kafka/logs'

server03

  1. version: '3'
  2. services:
  3. kafka-cluster:
  4. image: bitnami/kafka:3.2.0
  5. restart: always
  6. container_name: kafka-cluster
  7. user: "1001:0"
  8. ports:
  9. - 9092:9092
  10. - 9093:9093
  11. environment:
  12. - KAFKA_CFG_BROKER_ID=2
  13. - KAFKA_HEAP_OPTS=-Xmx1g -Xms1g
  14. - ALLOW_PLAINTEXT_LISTENER=yes
  15. - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
  16. - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
  17. - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka02.com.cn:9092
  18. - KAFKA_CFG_NUM_PARTITIONS=10
  19. - KAFKA_CFG_NUM_IO_THREADS=8
  20. - KAFKA_CFG_NUM_NETWORK_THREADS=3
  21. - KAFKA_ENABLE_KRAFT=yes
  22. - KAFKA_CFG_PROCESS_ROLES=broker,controller
  23. - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
  24. - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka01.com.cn:9093,2@kafka02.com.cn:9093,3@kafka03.com.cn:9093
  25. - KAFKA_KRAFT_CLUSTER_ID=1TGYcbFuRXa4Lqojs4B9Hw
  26. volumes:
  27. - "/etc/localtime:/etc/localtime"
  28. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/data:/bitnami/kafka/data'
  29. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/logs:/bitnami/kafka/logs'

 server04

  1. version: '3'
  2. services:
  3. kafka-cluster:
  4. image: bitnami/kafka:3.2.0
  5. restart: always
  6. container_name: kafka-cluster
  7. user: "1001:0"
  8. ports:
  9. - 9092:9092
  10. - 9093:9093
  11. environment:
  12. - KAFKA_CFG_BROKER_ID=3
  13. - KAFKA_HEAP_OPTS=-Xmx1g -Xms1g
  14. - ALLOW_PLAINTEXT_LISTENER=yes
  15. - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
  16. - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
  17. - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka03.com.cn:9092
  18. - KAFKA_CFG_NUM_PARTITIONS=10
  19. - KAFKA_CFG_NUM_IO_THREADS=8
  20. - KAFKA_CFG_NUM_NETWORK_THREADS=3
  21. - KAFKA_ENABLE_KRAFT=yes
  22. - KAFKA_CFG_PROCESS_ROLES=broker,controller
  23. - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
  24. - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka01.com.cn:9093,2@kafka02.com.cn:9093,3@kafka03.com.cn:9093
  25. - KAFKA_KRAFT_CLUSTER_ID=1TGYcbFuRXa4Lqojs4B9Hw
  26. volumes:
  27. - "/etc/localtime:/etc/localtime"
  28. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/data:/bitnami/kafka/data'
  29. - '/data/deploy/xdeas-docker/kafka-cluster/volumes/logs:/bitnami/kafka/logs'

docker-compose up -d

至此,zookeeper版本和Kraft版本都已搭建成功。请勿直接复制,各配置请根据个人情况稍作修改。 

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/小桥流水78/article/detail/822442
推荐阅读
相关标签
  

闽ICP备14008679号