Kafka一些命令操作和调优

# 创建 Topic：

kafka-topics.sh --create --zookeeper  localhost:2181 --replication-factor 3 --partitions 3 --topic test

# Topic 分区扩容

kafka-topics.sh --zookeeper  localhost:2181 --alter --topic test --partitions 4

# 删除 Topic：

kafka-topics.sh --delete --zookeeper  localhost:2181 localhost:9092 --topic test

#查询 Topic 详细信息

$ ./kafka-topics.sh --topic event_topic --zookeeper  localhost:2181 --describe
Topic:event_topic   PartitionCount:10   ReplicationFactor:2 Configs:compression.type=gzip
    Topic: event_topic  Partition: 0    Leader: 1001    Replicas: 1001,1003 Isr: 1001,1003
    Topic: event_topic  Partition: 1    Leader: 1003    Replicas: 1003,1002 Isr: 1003,1002
    Topic: event_topic  Partition: 2    Leader: 1002    Replicas: 1002,1001 Isr: 1002,1001
    Topic: event_topic  Partition: 3    Leader: 1001    Replicas: 1001,1002 Isr: 1001,1002
    Topic: event_topic  Partition: 4    Leader: 1003    Replicas: 1003,1001 Isr: 1003,1001
    Topic: event_topic  Partition: 5    Leader: 1002    Replicas: 1002,1003 Isr: 1002,1003
    Topic: event_topic  Partition: 6    Leader: 1001    Replicas: 1001,1003 Isr: 1001,1003
    Topic: event_topic  Partition: 7    Leader: 1003    Replicas: 1003,1002 Isr: 1003,1002
    Topic: event_topic  Partition: 8    Leader: 1002    Replicas: 1002,1001 Isr: 1002,1001
    Topic: event_topic  Partition: 9    Leader: 1001    Replicas: 1001,1002 Isr: 1001,1002

#列出全部 Topic

kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal

#查看group的消费情况

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092  --describe --group test-group

# 示例：
# TOPIC: group对应的topic
# PARTITION：aprtition编号，从0开始0-5表示有6个partition
# CURRENT-OFFSET：此消费着当前已消费的offset
# LOG-END-OFFSET：生产者在此partition分区上已提交确认的offset
# LAG：两个offset的差值，就是常说的积压。此数值过大为异常。
# HOST：消费者所在的服务器ip 
# CLIENT-ID：消费者的信息
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group  test-group

#删除group

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --delete --group test-group

#重新设置消费者位移

Earliest策略：把位移调整到当前最早位移处

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-earliest –execute

Latest策略：把位移调整到当前最新位移处

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-latest --execute
Current策略：把位移调整到当前最新提交位移处

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-current --execute
Specified-Offset策略：把位移调整到指定位移处

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-offset <offset> --execute
Shift-By-N策略：把位移调整到当前位移+N处(N可以是负值)

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group  --topic test --reset-offsets --shift-by <offset_N> --execute
DateTime策略：（把位移调整到大于给定时间的最小位移处）
时间需要减8
bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group   --topic test --reset-offsets --to-datetime 2019-06-20T20:00:00.000 --execute
Duration策略：把位移调整到距离当前时间指定间隔的位移处，然后将位移调整到距离当前给定时间间隔的位移处，具体格式是 PnDTnHnMnS。
    以字母 P 开头，后面由 4 部分组成，即 D、H、M 和 S，分别表示天、小时、分钟和秒。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --by-duration PT0H30M0S --execute

# 设置 topic 过期时间（单位 毫秒）

# 设置 topic 过期时间（单位 毫秒）
### 3600000 毫秒 = 1小时
./bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --entity-name topic-devops-elk-log-hechuan-huanbao --entity-type topics --add-config retention.ms=3600000

# 查看 topic 配置
./bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --describe --entity-name topic-devops-elk-log-hechuan-huanbao --entity-type topics

#使用脚本生产/消费消息

# 连接到test-topic，然后通过输入+会车生产消息
$ bin/kafka-console-producer.sh --broker-list kafka-host:port --topic test-topic  --producer-property
>

# --from-beginning: 指定从开始消费消息，否则会从最新的地方开始消费消息
$ bin/kafka-console-consumer.sh --bootstrap-server kafka-host:port --topic test-topic --group test-group --from-beginning --consumer-property

#kafka性能测试

# 测试生产者
# 向指定主题发送了 1 千万条消息，每条消息大小是 1KB
# 它会打印出测试生产者的吞吐量 (MB/s)、消息发送延时以及各种分位数下的延时
$ bin/kafka-producer-perf-test.sh --topic test-topic --num-records 10000000 --throughput -1 --record-size 1024 --producer-props bootstrap.servers=kafka-host:port acks=-1 linger.ms=2000 compression.type=lz4

2175479 records sent, 435095.8 records/sec (424.90 MB/sec), 131.1 ms avg latency, 681.0 ms max latency.
4190124 records sent, 838024.8 records/sec (818.38 MB/sec), 4.4 ms avg latency, 73.0 ms max latency.
10000000 records sent, 737463.126844 records/sec (720.18 MB/sec), 31.81 ms avg latency, 681.00 ms max latency, 4 ms 50th, 126 ms 95th, 604 ms 99th, 672 ms 99.9th.

# 测试消费者性能

$ bin/kafka-consumer-perf-test.sh --broker-list kafka-host:port --messages 10000000 --topic test-topic
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2019-06-26 15:24:18:138, 2019-06-26 15:24:23:805, 9765.6202, 1723.2434, 10000000, 1764602.0822, 16, 5651, 1728.1225, 1769598.3012

#磁盘目录优化

kafka 读写的单位是 partition，因此将一个 topic 拆分为多个 partition 可以提高吞吐量。但是这里有个前提，就是不同 partition 需要位于不同的磁盘（可以在同一个机器）。如果多个 partition 位于同一个磁盘，那么意味着有多个进程同时对一个磁盘的多个文件进行读写，使得操作系统会对磁盘读写进行频繁调度，也就是破坏了磁盘读写的连续性。

#JVM参数配置

1、推荐使用最新的 G1 来代替 CMS 作为垃圾回收器。推荐 Java 使用的最低版本为 JDK 1.7u51。
G1相比较于CMS的优势：
G1 是一种适用于服务器端的垃圾回收器，很好的平衡了吞吐量和响应能力
对于内存的划分方法不同，Eden, Survivor, Old 区域不再固定，使用内存会更高效。
G1 通过对内存进行 Region 的划分，有效避免了内存碎片问题。
G1 可以指定GC时可用于暂停线程的时间（不保证严格遵守）。
而 CMS 并不提供可控选项。
CMS 只有在 FullGC 之后会重新合并压缩内存，而G1把回收和合并集合在一起。
CMS 只能使用在 Old 区，在清理 Young 时一般是配合使用 ParNew，而 G1 可以统一两类分区的回收算法。
2、G1的适用场景：
JVM占用内存较大(At least 4G)
应用本身频繁申请、释放内存，进而产生大量内存碎片时。
对于GC时间较为敏感的应用。
目前我们使用的 JVM 参数：

#日志数据刷盘策略

为了大幅度提高 producer 写入吞吐量，需要定期批量写文件。

有 2 个参数可配置：

log.flush.interval.messages = 100000：

每当 producer 写入 100000 条数据时，就把数据刷到磁盘

log.flush.interval.ms=1000：

每隔 1 秒，就刷一次盘

#日志保留时间

当 kafka server 的被写入海量消息后，会生成很多数据文件，且占用大量磁盘空间，如果不及时清理，可能导致磁盘空间不够用，kafka 默认是保留7天。

参数：log.retention.hours = 168