Kafka2.8.x安装与配置

女爷i 2023-01-17 08:48 152阅读 0赞

### Kafka2.8.x安装与配置 ###

*  Kafka安装
 *   *  Kafka单机版安装
     *  Kafka常用操作命令
     *  Kafka集群安装及配置
 *  Kafka名词说明
 *  Kafka主题管理及分区
 *   *  主题管理
     *  分区管理
     *   *  副本机制
         *  分区Leader选举
         *  分区重新分配
         *   *  分区分配策略

# Kafka安装 #

参考官网：[http://kafka.apache.org/documentation/\#quickstart][http_kafka.apache.org_documentation_quickstart]  
前置条件要有JDK8+的环境，这里不赘述。

## Kafka单机版安装 ##

1、 解压Kakka安装包

# 1、 解压Kakka安装包
    cd /opt/kafka/
    tar -zxvf kafka_2.13-2.8.0.tgz
    cd kafka_2.13-2.8.0

2、配置zookeeper配置

#存放zk数据信息
    dataDir=/opt/kafka/kafka_2.13-2.8.0/data/zookeeper
    # the port at which the clients will connect
    clientPort=2181
    # disable the per-ip limit on the number of connections since this is a non-production config
    maxClientCnxns=0
    # Disable the adminserver by default to avoid port conflicts.
    # Set the port to something non-conflicting if choosing to enable this
    admin.enableServer=false
    # admin.serverPort=8080

3、配置kafka配置

less config/server.properties 
    #存放kafka broker节点ID，不同节点这个要唯一
    broker.id=0
    #主要配置
    listeners=PLAINTEXT://192.168.204.110:9092
    advertised.listeners=PLAINTEXT://192.168.204.110:9092
    log.dirs=/opt/kafka/kafka_2.13-2.8.0/data/kafka
    zookeeper.connect=localhost:2181

4、启动zk

./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

5、启动kafka服务

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

6、验证zk及kafka是否启动成功

#查看进程
    [root@localhost kafka_2.13-2.8.0]# jps -l
    14494 org.apache.zookeeper.server.quorum.QuorumPeerMain
    14958 jdk.jcmd/sun.tools.jps.Jps
    14879 kafka.Kafka
    #查看服务
    ps -ef|grep zookeeper
    ps -ef|grep kafka

## Kafka常用操作命令 ##

1、创建带分区副本的主题

# 创建一个名称为tp01的主题，2个分区，1个副本
    [root@localhost kafka_2.13-2.8.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic tp01 --partitions 2 --replication-factor 1
    Created topic tp01.

2、查看当前服务有多少主题

[root@localhost kafka_2.13-2.8.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --list
    __consumer_offsets
    test_topic
    tp01

3、查看主题详细信息

[root@localhost kafka_2.13-2.8.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic tp01
    Topic: tp01	TopicId: C8Ix8jEKSUmJLpV0cLvITA	PartitionCount: 2	ReplicationFactor: 1	Configs: 
    	Topic: tp01	Partition: 0	Leader: 0	Replicas: 0	Isr: 0
    	Topic: tp01	Partition: 1	Leader: 0	Replicas: 0	Isr: 0
    [root@localhost kafka_2.13-2.8.0]#

4、启动生产端发送消息

[root@localhost kafka_2.13-2.8.0]# bin/kafka-console-producer.sh --bootstrap-server 192.168.204.110:9092 --topic tp01
    >sada

5、启动消费端接收消息

[root@localhost kafka_2.13-2.8.0]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.204.110:9092 --topic tp01
    sada

## Kafka集群安装及配置 ##

后续更新

# Kafka名词说明 #

**Zookeeper：** zk负责维护和协调broker，当kafka系统中新增了broker或者某个broker发生故障失效时，由zk通知生产者和消费者。生产者和消费者依据zk和broker状态信息与broker协调数据的发布和订阅。  
**AR（Assigned Replicas）:** 分区中所有的副本统称AR  
**ISR（In-Sync Replicas）:** 所有与Leader部分保持一定程度的副本集合包括Leader副本。  
**OSR（Out-of-Sync-Replicas）:** 与leader副本同步滞后过多的副本。  
**HW（High Watermark）:** 高水位，标识了一个特定的offset。消费者只能拉取到这个offset之前的消息。  
**LEO(Log end Offset):** 即日志末端位移，记录了该副本底层日志中下一条消息，也就是说LEO=10，那么表示该副本保存了10条消息，位移值范围是【0-9】10为下一条消息的位移值。

# Kafka主题管理及分区 #

## 主题管理 ##

参考官网配置：[http://kafka.apache.org/documentation/\#topicconfigs][http_kafka.apache.org_documentation_topicconfigs]

1.  创建主题的命令：

参数说明：  
–zookeeper ：为kafka所配置的zk地址，多个zk以“，”号分隔开，用于存放kafka服务的主题元数据信息及节点信息。  
–partitions ：用于设置主题的分区数，每一个线程负责一个分区  
–replication-factor ：用于设置主题的副本数，每个副本分配在不同的节点，但不能超过总结点数，比如上面的示例若副本设置为2就会报错。

1.  修改主题配置：

# 为主题增加配置
    [root@localhost bin]# ./kafka-topics.sh --alter --zookeeper localhost:2181 --topic tp01 --config flush.messages=1
    WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases.
             Going forward, please use kafka-configs.sh for this functionality
    Updated config for topic tp01.
    
    # 删除刚才新增的配置
    [root@localhost bin]# ./kafka-topics.sh --alter --zookeeper localhost:2181 --topic tp01 --delete-config flush.messages
    WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases.
             Going forward, please use kafka-configs.sh for this functionality
    Updated config for topic tp01.

1.  删除主题：
    
     *  若delete.topic.enable=true  
        直接彻底删除该Topic。
     *  若delete.topic.enable=false  
        若当前被删除的topic没有使用过即没有传输过信息，可以彻底删除； 反之，则不会把这个topic直接删除，而是将其标记为（marked for deletion），重启kafka server后删除。

命令：

[root@localhost bin]# ./kafka-topics.sh --delete --zookeeper localhost:2181 --topic test_topic
    Topic test_topic is marked for deletion.

## 分区管理 ##

Kafka可以将主题分为多个分区（Partitions），会根据分区规则选择把消息存储在那个分区中，只要如果分区规则设置合理，那么所有的消息将会被均匀的分不到不同的分区中，这样就实现了负载均衡和水平扩展，另外多个订阅者可以从一个或者多个分区中同时消费数据，以支撑海量数据处理能力。  
由于消息是以追加到分区中的，多个分区顺序写磁盘的总效率要比随机写内存还要高，是kafka高吞吐的重要保证之一。

1.  增加分区数（分区数只能修改新增不能减少个数）

[root@localhost kafka_2.13-2.8.0]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --topic tp01 --partitions 5
    WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
    Adding partitions succeeded!
    [root@localhost kafka_2.13-2.8.0]#

### 副本机制 ###

由于Producer和Consumer都会与leader角色的分区副本相连，所以kafka需要以集群的组织形式提供主题下的消息高可用，kafka支持主从复制，所以消息具备高可用和持久性。  
通常情况下，增加分区可以提供kafka集群的吞吐量。然后，而应该意识到集群的总分区数或者单台服务器上的分区数过多，会增加不可用及延迟的风险。

### 分区Leader选举 ###

如果某个分区的leader挂了，那么其它跟随着将会进行选举产生一个新的leader，之后所有的读写就会转移到新的Leader上，在kafka中，其不是采用常见的多数选举方式进行副本的Leader选举，而是会在ZK上针对每个主题维护一个称为ISR（in-sync replica，已同步的副本）的集合，显然还有一些副本没有来得及同步。只有这个ISR列表里面的才有资格称为leader（先使用ISR里面的第一个，如果不行以此类推，因为ISR里面的是同步副本，消息是最完整且各个节点都是一样的）。  
通过ISR，kafka需要冗余度较低，可以容忍的失败数比较高，假设某个topic有f+1个副本，kafka可以容忍f个不可以，当然，如果全部ISR里面的副本都不可用，也可以选择其它可用的副本，只是存在的数据不一致。

### 分区重新分配 ###

正常业务场景中对于kafka集群里面的添加机器是正常的需求，而且添加起来非常方便，我们需要做的事是从已经部署好的kafka节点中复制响应的配置文件，然后把里面的broker id修改成全局唯一的，最后启动这个节点，即可将它假如到现有的kafka集群中。

但是新加的kafka节点服务器不会自动地分配已存在的数据或分区，所以无法分担集群的负载工作，除非我们新建一个topic，但是我们想手动将部分分区移到新添加的kafka节点上，kafka内部提供了相关的工具来重新分布某个topic的分区。  
重新分配场景：

比如我们最初由3个节点4个分区3个副本，那么必有一个节点服务承载2个分区的数据，这样就造成了数据分配不均衡，所以我们会新增一个节点，但是新增的节点不会去承载原有的分区数据，也就是仍然是三个节点承载4个分区数据，所以我们要借助kafka-reassign-partitions.sh工具生成reassign plan。

1.  先按照要求定义一个文件，里面说明那些topic需要重新分区，内容格式如下：

vi reassign.json
    { 
    "topics":[{ "topic":"tp01"}],
    "version":1
    }
    :wq保存

1.  使用**kafka-reassign-partitions.sh**工具生成reassgin plan

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file reassign.json --broker-list "brokerid0,1,2.." --generate

执行上述命令后，会生成俩个JSON字符串：，第一个JSON内容为当前的分区副本分配情况，第二个为重新分配的候选方案，注意这里只是生成一份可行性方案，并没有真正执行重新分配动作。  
我们将第二个JSON内容保存到名为result.json文件里面（文件名不重要，文件格式也不一定是json为结尾，只要保证内容是JSON即可），然后执行这些reassign plan.  
3. 执行分配策略

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file result.json --execute

1.  查看当前重新分配的进度

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file result.json --verify

#### 分区分配策略 ####

**RangeAssignor 分配策略**  
此原理是按照消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能的均匀分配给所有的消费者，对于每一个主题，此策略会将消费组内所有订阅这个topic的消费者按照名称的字典序排序，然后为每个消费者划分固定的分区范围，如果不够平均分配，那么字典序考前的消费者会被多分配一个分区。

假设n=分区数/消费者数量，m=分区数%消费者数量，那么前m个消费者每个分配n+1个分区，第m个消费者分配n个分区

例如：2个主题3个分区，那么所订阅的所有分区标识为，t0p0、t0p1、t0p2、t1p0、t1p1、t1p2、  
那么若俩个消费者C1\\C2 n=3/2=1 m=1  
c1 分配：t0p0、t1p0、t1p0、t1p1  
c2 分配：t0p2、t1p2  
这样可以明显看到这样的分配并不均匀，如果类似情形扩大，就会造成消费者过载的现象。  
**RoundRobinAssignor 分配策略**  
该策略是将消费组内所有的消费组及消费者锁定的所有topic的分区按照字典序排序，然后通过轮询的方式逐个将分区以此分配给每个消费者。对于上面的示例用词策略分配为下：  
c1 分配：t0p0、t0p2、t1p1  
c2 分配：t0p1、t1p0、t1p2

但是如果同一个消费组内的消费者所订阅的信息是不同的，那么在执行分区分配的时候就是不完全的轮询分配，有可能会导致的分区分配不均匀，如果某个消费者没有订阅消费组内的某个topic，那么在分配分区的时候此消费者将不会分配这个主题对应的任何分区。

示例：  
假设消费组有三个消费者C0,C1,C2他们共订阅了3个主题：T0,T1,T2,这三个主题分别有1,2,3个分区，  
即整个消费组订阅了T0P0, T1P0, T1P1, T2P0, T2P1,T2P2共6个分区，具体而言就是消费组C0订阅是主题T0，消费组C1订阅了主题T0,T1,消费者C2订阅了主题T0,T1,T2,那么最终分配的结果：  
C0: t0p0  
C1: t1p0  
C2: t1p1、t2p0，t2p1，t2p2  
可以看到此策略并不是最优的分案

**StickyAssignor分配策略：**  
分区的分配要可能的均匀；  
分区的分配尽可能的与上次分配的保持相同。

当倆者发生冲突时，第一个目标优于第二个目标。

[http_kafka.apache.org_documentation_quickstart]: http://kafka.apache.org/documentation/#quickstart
[http_kafka.apache.org_documentation_topicconfigs]: http://kafka.apache.org/documentation/#topicconfigs