当前位置:   article > 正文

kafka常用脚本_kafka脚本

kafka脚本

kafka相关的脚本

kafka脚本

kafka启动,可写成如下脚本

## 启动:kafka.sh start

#! /bin/bash
case $1 in
"start"){
 for i in node04 node05 node06
 do
 echo " --------启动 $i Kafka-------"
 ssh $i "source /etc/profile ; /opt/software/kafka/bin/kafka-server-start.sh -daemon /opt/software/kafka/config/server.properties"
 done
};;
"stop"){
 for i in node04 node05 node06
 do
 echo " --------停止 $i Kafka-------"
 ssh $i "source /etc/profile; /opt/software/kafka/bin/kafka-server-stop.sh  "
 done
};;
esac
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
列举topic:	
/bin/kafka-topics.sh --bootstrap-server node05:9092 --list
dsp_bid
dsp_click
dsp_monitor 

###新建topic 
/bin/kafka-topics.sh --bootstrap-server node05:9092 --create  --replication-factor 2 --partitions 3 --topic t_monitor

###查看topic
/bin/kafka-topics.sh --bootstrap-server node05:9092 --describe --topic t_monitor
###  Topic: t_monitor  Partition: 0   Leader: 368   Replicas: 368,370,369  Isr: 368,370,369

###生产者
/kafka/bin/kafka-console-producer.sh --bootstrap-server node05:9092 --topic t_monitor

## 消费topic数据
/bin/kafka-console-consumer --bootstrap-server hostname:9092 --topic t_monitor --from-beginning 

##删除topic
 bin/kafka-topics.sh --bootstrap-server hostname:9092 --delete --topic first

##修改topic的分区数(只能增加不能减少)
bin/kafka-topics.sh --bootstrap-server hostname:9092 --alter --topic first --partitions 3





##查看消费者组的列表
bin/kafka-consumer-groups.sh --bootstrap-server node01:9092 --list

##查看消费组的消费详情
bin/kafka-consumer-groups.sh  --bootstrap-server node01:9092 --group test-01-group --describe 

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

1. kafka生产者

1. 生产者生产数据过程如下:
kafka生产数据过程
buffer.memory :RecordAccumulator 缓冲区总大小,默认 32m
batch.size :缓冲区一批数据最大值,默认 16k。适当增加该值,可以提高吞吐量,但是如果该值设置太大,会导致数据传输延迟增加
linger.ms :如果数据迟迟未达到 batch.size,sender 等待 linger.time之后就会发送数据。单位 ms,默认值是 0ms,表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
compression.type :生产者发送的所有数据的压缩方式。默认是 none,也就是不压缩。支持压缩类型:none、gzip、snappy、lz4 和 zstd

2. 生产者发送数据分区策略
(1)指明partition的情况下,直接将指明的值作为partition值;例如partition=0,所有数据写入分区0
(2)没有指明partition值但有key的情况下,将key的hash值与topic的partition数进行取余得到partition值;
例如:key1的hash值=5, key2的hash值=6 ,topic的partition数=2,那 么key1 对应的value1写入1号分区,key2对应的value2写入0号分区。
(3)既没有partition值又没有key值的情况下,Kafka采用Sticky Partition(黏性分区器),会随机选择一个分区,并尽可能一直使用该分区,待该分区的batch已满或者已完成,Kafka再随机一个分区进行使用(和上一次的分区不同)。
例如:第一次随机选择0号分区,等0号分区当前批次满了(默认16k)或者linger.ms设置的时间到, Kafka再随机一个分区进
行使用(如果还是0会继续随机)。
(4)自定义分区器:实现Partitioner接口,实现 3 个方法:partition,close,configure 编写 partition 方法,返回分区号

3.生产者发送数据ACK应答级别
acks=0,生产者发送过来数据就不管了,可靠性差,效率高;
acks=1,生产者发送过来数据Leader应答,可靠性中等,效率中等;
acks=-1,生产者发送过来数据Leader和ISR队列里面所有Follwer应答,可靠性高,效率低;
在生产环境中,acks=0很少使用;acks=1,一般用于传输普通日志,允许丢个别数据;acks=-1,一般用于传输和钱相关的数据,
对可靠性要求比较高的场景。

properties.put(ProducerConfig.ACKS_CONFIG, "all"); --也是-1
  • 1

2. Kafka Broker

1. Kafka Broker 总体工作流程
kafka broker工作流程

2. Broker 重要参数

##ISR 中,如果 Follower 长时间未向 Leader 发送通信请求或同步数据,则该 Follower 将被踢出 ISR。该时间阈值,默认 30s。
replica.lag.time.max.ms

##默认是 true。 自动 Leader Partition 平衡。
auto.leader.rebalance.enable 

##默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值,控制器会触发 leader 的平衡。
leader.imbalance.per.broker.percentage 

##默认值 300 秒。检查 leader 负载是否平衡的间隔时间。
leader.imbalance.check.interval.seconds 

##Kafka 中 log 日志是分成一块块存储的,此配置是指 log 日志划分 成块的大小,默认值 1G。
log.segment.bytes 

##默认 4kb,kafka 里面每当写入了 4kb 大小的日志(.log),然后就往 index 文件里面记录一个索引。
log.index.interval.bytes 

##中数据保存的时间,默认 7 天。
log.retention.hours Kafka 

##中数据保存的时间,分钟级别,默认关闭。
log.retention.minutes Kafka 

##Kafka 中数据保存的时间,毫秒级别,默认关闭。
log.retention.ms 

## 检查数据是否保存超时的间隔,默认是 5 分钟。
log.retention.check.interval.ms

##默认等于-1,表示无穷大。超过设置的所有日志总大小,删除最早的 segment。
log.retention.bytes 

##默认是 delete,表示所有数据启用删除策略;如果设置值为 compact,表示所有数据启用压缩策略。
log.cleanup.policy 

##默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。 num.replica.fetchers 副本拉取线程数,这个参数占总核数的 50%1/3
num.io.threads 

##默认是 3。数据传输线程数,这个参数占总核数的50%2/3 。
num.network.threads 

## 强制页缓存刷写到磁盘的条数,默认是 long 的最大值,9223372036854775807。一般不建议修改,交给系统自己管理。
log.flush.interval.messages

##每隔多久,刷数据到磁盘,默认是 null。一般不建议修改,交给系统自己管理。
log.flush.interval.ms 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47

3. kafka消费者再平衡

## Kafka 消费者和 coordinator 之间的心跳时间,默认 3s。该条目的值必须小于 session.timeout.ms,也不应该高于session.timeout.ms 的 1/3。
heartbeat.interval.ms : 3

## Kafka 消费者和 coordinator 之间连接超时时间,默认 45s。超过该值,该消费者被移除,消费者组执行再平衡。
session.timeout.ms :45

##消费者处理消息的最大时长,默认是 5 分钟。超过该值,该消费者被移除,消费者组执行再平衡。
max.poll.interval.ms : 5min

## 消费者分区策略 : Range 、 RoundRobin 、 Sticky 、CooperativeSticky
partition.assignment.strategy : Range + CooperativeSticky

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

若:消费者0挂

range模式:
消费者组需要按照超时时间 45s 来判断它是否退出,所以需要等待,时间到了 45s 后,判断它真的退出就会把任务分配给其他 broker 执行。消费者 0 已经被踢出消费者组,所以重新按照 range 方式分配。

RoundRobin :模式
0 号消费者挂掉后,消费者组需要按照超时时间 45s 来判断它是否退出,所以需要等待,时间到了 45s 后,判断它真的退出就会把任务分配给其他 broker 执行。消费者 0 已经被踢出消费者组,所以重新按照 RoundRobin 方式分配

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

闽ICP备14008679号