# 创建 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