Kafka实践

た入场券 2022-02-21 09:39 229阅读 0赞

以下命令示范在kafka版本为0.10.1.0下进行。

### \* 查看kafka版本 ###

kafka没有提供version命令，不确定是否有方便的方法，但你可以进入kafka/libs文件夹。你应该看到像kafka\_2.11-0.10.1.0.jar这样的文件，其中2.11是Scala版本，0.10.1.0是Kafka版本。

[链文][Link 1]

### \* 启动 ###

Kafka使用ZooKeeper，因此如果您还没有ZooKeeper服务器，则需要先启动它。 您可以使用与kafka一起打包的便捷脚本来获得快速且简单的单节点ZooKeeper实例。

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

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

然后:

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

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

集群:  
    
  zookeeper集群化：    
  ①：    
  在zookeeper.properties中添加相关配置(在各服务器对应的dataDir目录下新建myid，值分别对应server.x)  
   server.1=xx.xx.xx.30:2888:3888  
   server.2=xx.xx.xx.31:2888:3888  
   server.3=xx.xx.xx.32:2888:3888  
     
  ②

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

kafka集群化：

①：

cp config/server.properties config/server-1.properties
       cp config/server.properties config/server-2.properties

编辑以下两个文件:  
     
   config/server-1.properties:  
    broker.id=1  
    listeners=PLAINTEXT://:9093  
    log.dirs=/tmp/kafka-logs-1  
   
   config/server-2.properties:  
    broker.id=2  
    listeners=PLAINTEXT://:9094  
    log.dirs=/tmp/kafka-logs-2  
    
  ②：

> bin/zookeeper-server-start.sh config/zookeeper.properties
      > bin/kafka-server-start.sh config/server.properties &
      > bin/kafka-server-start.sh config/server-1.properties &   
      > bin/kafka-server-start.sh config/server-2.properties &

[官网][Link 2]   or  [其它][Link 3]

### **\* 停止** ###

bin/kafka-server-stop.sh
        bin/zookeeper-server-stop.sh

### \* 配置 ###

\* kafka0.8版本前后  
   0.8 以前，消费进度是直接写到 zookeeper 的，consumer 必须知道 zookeeper 的地址。这个方案有性能问题，0.9 的时候整体大改了一次，brokers 接管了消费进度，consumer 不再需要和 zookeeper 通信了。

###    \* zookeeper在kafka的作用 ###

Apache Kafka的一个关键依赖是Apache Zookeeper，它是一个分布式配置和同步服务。 Zookeeper是Kafka代理和消费者之间的协调接口。 Kafka服务器通过Zookeeper集群共享信息。 Kafka在Zookeeper中存储基本元数据，例如关于主题，代理，消费者偏移(队列读取器)等的信息。  
       由于所有关键信息存储在Zookeeper中，并且它通常在其整体上复制此数据，因此Kafka代理/ Zookeeper的故障不会影响Kafka集群的状态。 Kafka将恢复状态，一旦Zookeeper重新启动。 这为Kafka带来了零停机时间。       Kafka代理之间的领导者选举也通过使用Zookeeper在领导者失败的情况下完成。

\* Kafka中bootstrap-server、broker-list和zookeeper的区别

broker-list: 对于console生产者，--broker-list参数指定了所使用的broker;  
  zookeeper：对于消费者，对于旧版本kafka的消费者，由--zookeeper参数设置,consumer的信息将会存放在zk之中;  
  bootstrap-server：对于消费者，对于新版本kafka的消费者，由--bootstrap-server参数设置,consumer的信息将会存放在kafka之中

[kafka工作流程][kafka]

### \* 查看现在topic列表 ###

bin/kafka-topics.sh --list --zookeeper localhost:2181  
   (在笔者作用版本下，官网的 bin/kafka-topics.sh --list --bootstrap-server localhost:9092  命令无法正常执行，报: bootstrap-server is not a recognized option,另换为broker-list,同样报:broker-list is not a recognized option)     
    猜测上述原因为构建的集群形式为zookeeper形式,因为按照正常情况下0.8之后的版本用bootstrap-server应该都正常，用zookeeper报上面这个异常才对，真怀疑现在用的kafka实际是0.8版本以下。

### \* 查看现在consumer group列表，使用--list参数 ###

bin/kafka-consumer-groups.sh --zookeeper 127.0.0.1:2181 --list

[Kafka查看topic、consumer group状态命令][Kafka_topic_consumer group]

### \* 查看特定consumer group 详情，使用--group与--describe参数 ###

bin/kafka-consumer-groups.sh --zookeeper 127.0.0.1:2181 --group console-consumer-708 --describe

GROUP                          TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER  
   console-consumer-708           bjjs-framework                 0          0               0               0               none

其中依次展示group名称、消费的topic名称、partition id、consumer group最后一次提交的offset、最后提交的生产消息offset、消费offset与生产offset之间的差值、当前消费topic-partition的group成员id(不一定包含hostname)

###       \* 创建topic ###

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic rz_test

注： partitions指定topic分区数，replication-factor指定topic每个分区的副本数

**partitions分区数:**  
        partitions ：分区数，控制topic将分片成多少个log。可以显示指定，如果不指定则会使用broker(server.properties)中的num.partitions配置的数量  
        虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗。  
        分区数也限制了consumer的并行度，即限制了并行consumer消息的线程数不能大于分区数  
        分区数也限制了producer发送消息是指定的分区。如创建topic时分区设置为1，producer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。  
          
   ** replication-factor副本:**  
        replication factor 控制消息保存在几个broker(服务器)上，一般情况下等于broker的个数。  
        如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量  
          
   [Kafka Shell基本命令（包括topic的增删改查）][Kafka Shell_topic]

### \* 控制台向topic生产数据 ###

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic rz_test

### \* 控制台消费topic的数据 ###

从最新offset处消费:

bin/kafka-console-consumer.sh  --zookeeper localhost:2181  --topic rz_test

从kafka保存数据的最起始地方处消费(默认保留数据7天):

# 此命令根据实际需要增删参数
      bin/kafka-console-consumer.sh  --zookeeper localhost:2181  --topic rz_test --consumer-property group.id=old-consumer-test --consumer-property consumer.id=old-consumer-cl --from-beginning --delete-consumer-offsets

consumer-property 参数以键值对的形式指定消费者级别的配置.  
  from-beginning 设置消息起始位置开始消费。默认是从新位置 latest开始消费.  
  delete-consumer-offsets 删除在zookeeper中记录已消费的偏移量.  
    
         旧版消费者默认将消费偏移量保存到zookeeper中，可以通过offsets.storage=kafka设置offsets.storage=kafka，则保存到kafka主题中,offsets.storage=zookeeper, 则保存到zookeeper中。

[kafka consumer消费者 offset groupID详解][kafka consumer_ offset groupID]

[kafka消费者基本操作][kafka 1]

### \* 查看topic某分区偏移量最大（小）值 ###

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic rz_test --time -1 --broker-list localhost:9092 --partitions 0

rz\_test:0:8

topic为rz\_test，分区为0的偏移量最大为8，即消费了8条数据(位置)  
    
  注： time为-1时表示最大值，time为-2时表示最小值

### \* 设置consumer group的offset ###

确保在这之前停止所有消费者，否则他们可能会这样做覆盖你写的补偿。

1. 启动zookeeper client

/zookeeper/bin/zkCli.sh

2. 通过下面命令设置consumer group:DynamicRangeGroup topic:DynamicRange partition:0的offset为1288:

set /consumers/DynamicRangeGroup/offsets/DynamicRange/0 1288

3. 注意如果你的kafka设置了zookeeper root，比如为/kafka，那么命令应该改为：

set /kafka/consumers/DynamicRangeGroup/offsets/DynamicRange/0 1288

生效:  重启相关的应用程序，就可以从设置的offset开始读数据了。

[kafka的offset的重置][kafka_offset]

手动更新Kafka存在Zookeeper中的偏移量。我们有时候需要手动将某个主题的偏移量设置成某个值，这时候我们就需要更新Zookeeper中的数据了。Kafka内置为我们提供了修改偏移量的类：kafka.tools.UpdateOffsetsInZK，我们可以通过它修改Zookeeper中某个主题的偏移量，具体操作如下：

[www@kafka ~]$ ./bin/kafka-run-class.sh kafka.tools.UpdateOffsetsInZK
    USAGE: kafka.tools.UpdateOffsetsInZK$ [earliest | latest] consumer.properties topic

在不输入参数的情况下，我们可以得知kafka.tools.UpdateOffsetsInZK类需要输入的参数。  
这个工具只能把Zookeeper中偏移量设置成earliest或者latest，如下：

[www@kafka ~]$ ./bin/kafka-run-class.sh kafka.tools.UpdateOffsetsInZK  earliest config/consumer.properties connect-test
     　　　　
    updating partition 0 with new offset: 8
    updated the offset for 1 partitions

\* 查看topic消费进度

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group testgroup --topic rz_test --zookeeper 127.0.0.1:2181

GROUP     TOPIC      PID                 OFFSET                   LOGSIZE   LAG

消费者组   topic名字  partition id    当前已消费的条数   总条数        未消费的条数

[Kafka常用命令使用说明][Kafka]

###    \* 查看指定topic信息           ###

bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic rz_test

Topic:rz\_test   PartitionCount:1        ReplicationFactor:1     Configs:  
            Topic: rz\_test  Partition: 0    Leader: 2       Replicas: 2     Isr: 2  
    
    第一行显示partitions的概况，列出了Topic名字，partition总数，存储这些partition的broker数

以后每一行都是其中一个partition的详细信息：

**leader **

是该partitons所在的所有broker中担任leader的broker id，每个broker都有可能成为leader

**replicas **

显示该partiton所有副本所在的broker列表，包括leader，不管该broker是否是存活，不管是否和leader保持了同步。

**isr **

in-sync replicas的简写，表示存活且副本都已同步的的broker集合，是replicas的子集  
      
    [Kafka常用命令使用说明][Kafka]

###    \* 查看指定topic所在group ###

###    \* 删除topic(慎用) ###

① 删除kafka的topic(只会删除zookeeper中的元数据，消息文件须手动删除)

bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic rz_test1

如果server.properties中没有把delete.topic.enable设为true，那么此时的删除并不是真正的删除，而是把topic标记为：marked for deletion

② 删除kafka中该topic相关的目录  
     
     在server.properties中找到配置log.dirs，把该目录下test0相关的目录删掉  
       
   ③ 登录zookeeper client。

/home/ZooKeeper/bin/zkCli.sh

④ 删除zookeeper中该topic相关的目录

rm -r /kafka/config/topics/test0

rm -r /kafka/brokers/topics/test0

rm -r /kafka/admin/delete\_topics/test0 （topic被标记为marked for deletion时需要这个命令）

⑤重启zookeeper和broker

\* 修改topic的partition数量（只能增加不能减少）

bin/kafka-topics.sh --alter --zookeeper 127.0.0.1:2181 --partitions 2 --topic rz_test

[Kafka常用命令使用说明][Kafka]

###    \* 修改topic的副本数 ###

① 自己写一个文件addReplicas.json，文件的内容是JSON格式的，比如：

{
           "version": 1,
           "partitions": [
               {
                   "topic": "test0",
                   "partition": 0,
                   "replicas": [
                        1,2
                   ]
               },
               {
                   "topic": "test0",
                   "partition": 1,
                   "replicas": [
                        1,2,3
                   ]
               },
               {
                   "topic": "test0",
                   "partition": 2,
                   "replicas": [
                        1,2,3
                   ]
               }
            ]
        }

② 运行命令：

bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:9092 --reassignment-json-file addReplicas.json --execute

### \* 下线broker ###

比如下线broker0

bin/kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 127.0.0.1:2181 --broker 0 --num.retries 3 --retry.interval.ms 60

shutdown broker

如果需要迁移，需要其他额外的操作

###    \* 生产消息的模型 ###

*  At most once 消息可能会丢，但绝不会重复传输
 *  At least one 消息绝不会丢，但可能会重复传输
 *  Exactly once 每条消息肯定会被传输一次且仅传输一次，很多时候这是用户所想要的。

[   Kafka delivery保证(kafka消息投递保证)][_Kafka delivery_kafka]

###       \* 消费消息的几种方式 ###

最新(lastest)  
   开头(earliest or beginning)  
   指定offset  
     
[kafka consumer消费者 offset groupID详解][kafka consumer_ offset groupID]

###    \* 生产者生产过程缺失数据、消费者重复消费数据的原因和解决方案 ###

[Kafka学习之怎么保证不丢，不重复消费数据][Kafka 1]

[Kafka重复消费和丢失数据研究][Kafka 2]

[kafka学习总结（处理多个consumer只消费topic数据一次）][kafka_consumer_topic]

[kafka中partition和消费者对应关系][kafka_partition]

###    \* 需要重启的操作和不需要重启的操作 ###

需要重启:

1. 配置文件变更内容

不需要重启:

1. 配置文件内容以命令形式变更，但若kafka服务重启则还是按照配置文件中的内容执行(比如：[kafka修改日志保留时间参数][kafka 2]) 。

2. topic、partition的变更。

[Link 1]: https://blog.csdn.net/Damonhaus/article/details/54310868
[Link 2]: http://kafka.apache.org/quickstart
[Link 3]: http://www.54tianzhisheng.cn/2018/01/04/Kafka/
[kafka]: https://www.w3cschool.cn/apache_kafka/apache_kafka_workflow.html
[Kafka_topic_consumer group]: https://blog.csdn.net/qq_40633152/article/details/81450910
[Kafka Shell_topic]: https://www.cnblogs.com/xiaodf/p/6093261.html
[kafka consumer_ offset groupID]: https://blog.csdn.net/suphieliang/article/details/80421577
[kafka 1]: http://www.cnblogs.com/sunshine-long/p/9857107.html
[kafka_offset]: https://www.cnblogs.com/hd-zg/p/5831219.html
[Kafka]: https://blog.csdn.net/lkforce/article/details/77776684
[_Kafka delivery_kafka]: https://blog.csdn.net/lizhitao/article/details/44199257
[Kafka 1]: https://blog.csdn.net/u010627840/article/details/76435385
[Kafka 2]: https://blog.csdn.net/zollty/article/details/53958641
[kafka_consumer_topic]: https://blog.csdn.net/zhaishujie/article/details/71713794
[kafka_partition]: https://blog.csdn.net/u013939918/article/details/70649764
[kafka 2]: https://blog.csdn.net/uuilghiugiu/article/details/78191844