Kafka快速上手

# 一、单机单broker快速上手
```
$ 解压并进入目录
tar -xzf kafka_2.11-2.3.0.tgz
mv kafka_2.11-2.3.0 /opt/kafka
cd /opt/kafka

$ 启动zookeeper
nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties &

$ 启动kafka
nohup ./bin/kafka-server-start.sh config/server.properties &

$ 创建blogTopic话题
./bin/kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 1 --partitions 1 --topic blogTopic
Created topic blogTopic.

$ 查看Topic列表。可以配置broker中发布一个不存在的topic时自动创建topic。
./bin/kafka-topics.sh --zookeeper localhost:2181 --list
blogTopic

$ 查看Topic详情
./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic blogTopic
Topic:blogTopic  PartitionCount:1    ReplicationFactor:1 Configs:
        Topic: blogTopic Partition: 0    Leader: 0   Replicas: 0 Isr: 0

$ Kafka提供了一个生成消息的工具，可以从输入文件或者命令行中读取消息并发送给Kafka集群，每一行是一条消息。
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic blogTopic
>add a blog
>add another blog

$ Kafka提供了一个消费消息的工具，可以消费指定的topic。
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic blogTopic

```

# 二、单机多broker快速上手
在一台机器上跑多个broker。
```
$ 解压并进入目录
tar -xzf kafka_2.11-2.3.0.tgz
mv kafka_2.11-2.3.0 /opt/kafka
cd /opt/kafka

$ 准备三个broker的配置文件
cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties
cp config/server.properties config/server-3.properties

$ 分别修改三个配置文件
vim config/server-1.properties
broker.id=1
listeners=PLAINTEXT://:9092
log.dirs=/tmp/kafka-logs-1

vim config/server-2.properties
broker.id=2
listeners=PLAINTEXT://:9093
log.dirs=/tmp/kafka-logs-2

vim config/server-3.properties
broker.id=3
listeners=PLAINTEXT://:9094
log.dirs=/tmp/kafka-logs-3

$ 启动zookeeper
nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties &

$ 启动kafka
nohup ./bin/kafka-server-start.sh config/server-1.properties &
nohup ./bin/kafka-server-start.sh config/server-2.properties &
nohup ./bin/kafka-server-start.sh config/server-3.properties &

$ 创建 articleTopic 话题，包含1个分区，3个副本。
./bin/kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 3 --partitions 1 --topic articleTopic
Created topic articleTopic.

$ 查看topic详情
./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic articleTopic
Topic:articleTopic      PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: articleTopic     Partition: 0    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
第一行是所有分区的摘要，其次，每一行提供一个分区信息，因为我们只有一个分区，所以只有一行。
1.Partition:  分区编号。
2.Leader：    leader所在broker的id，该节点负责该分区的所有的读和写，每个节点的 leader 都是随机选择的。
3.Replicas：  备份的节点列表id，无论该节点是否是 leader 或者目前是否还活着，不管node死活，只是列出信息而已。
4.Isr：       同步备份的节点列表id，也就是活着的节点并且正在同步 leader。

$ 生产消息，然后消费消息
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic articleTopic
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic articleTopic

$ 杀掉节点2，查看Topic详情，可以看到，Leader从2变成了3，Isr列表中少了2。
ps aux | grep server-2.properties | awk '{print $2}' |xargs kill
./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic articleTopic
Topic:articleTopic      PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: articleTopic     Partition: 0    Leader: 3       Replicas: 2,3,1 Isr: 3,1

$ 节点2死掉以后消费消息，看是否成功。
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic articleTopic

```

# 三、常用命令总结

- `partitions`:
 - `partitions` ：指定topic分区数，控制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`配置的数量

```
$ 创建topic
./bin/kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 3 --partitions 5 --topic articleTopic

$ 查看topic列表
./bin/kafka-topics.sh --zookeeper localhost:2181 --list

$ 查看topic信息
./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic articleTopic

$ 生产消息
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic articleTopic

$ 消费消息
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic articleTopic

$ 查看某分区偏移量。查看`topic`某分区偏移量最大值、最小值。time为-1时表示最大值，time为-2时表示最小值。
./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic articleTopic --partitions 0 --time -1

$ 修改topic分区数。增加分区数为10个。
./bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic articleTopic --partitions 10

$ 修改topic的保留时间，修改为86400000
./bin/kafka-topics.sh --zookeeper server1:2181 --alter --topic flowai_0509 --config retention.ms=86400000 --config cleanup.policy=delete

$ 删除topic。这只会被标记删除，实际文件并未删除。
./bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic articleTopic

$ 通过组消费。`articleGroup`会被自动创建。
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic articleTopic -group articleGroup --from-beginning

$ 查看组列表
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

$ 查看组详情
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group articleGroup --describe

```

余子越的博客

最近发表

友情链接