【Kafka 技术专题】「实践操作篇」单机部署实践手册（2.8.0）

下载 kafka

wget https://archive.apache.org/dist/kafka/0.8.0/kafka_2.8.0-0.8.0.tar.gz

或者

curl -O https://archive.apache.org/dist/kafka/0.8.0/kafka_2.8.0-0.8.0.tar.gz

解压缩

 tar -vxf kafka_2.8.0-0.8.0.tar.gz

修改配置

修改 conf/server.properties

• 【host.name=10.10.224.12】：修改为主机 ip,不然服务器返回给客户端的是主机的 hostname,客户端并不一定能够识别。

修改 conf/zookeeper.properties

• 【dataDir=/usr/local/tmp/zookeeper】：zookeeper 临时数据文件。

启动 zookeeper 和 kafka

启动 zookeeper

• 【./zookeeper-server-start.sh ../config/zookeeper.properties &】(&推出命令行，服务守护执行)

启动 kafka

• 【./kafka-server-start.sh ../config/server.properties &】：启动相关的 kafka 服务，指定相关的配置文件

验证是否成功

创建主题

• 不过我们一般不建议将 auto.create.topics.enable 参数设置为 true，因为这个参数会影响 topic 的管理与维护。

• kafka 的 0.10 版本之前，可以使用内置的 kafka-admin 包，后续提供了专门的类 AdminClient API 来进行 API 层面上的 topic 创建。

• 通过 kafka 提供的 kafka-topics.sh 脚本来创建，并且我们也建议通过这种方式（或者相关的变种方式）来创建 topic。

./kafka-create-topic.sh --partition 1 --replica 1 --zookeeper localhost:2181 --topic test

或者 低版本

bin/kafka-topics.sh --create --zookeeper 192.168.0.2:2181/kafka100 --topic topic-test1 --replication-factor 2 --partitions 4

• 2181：是 zookeeper 端口

• partion：topic 的数据内容被划分为几块存储

• replication-factor: 物理存储 topic 的内容采用几个副本的容错策略

自动创建主题

如果 kafka broker 中的 config/server.properties 配置文件中配置了 auto.create.topics.enable 参数为 true（默认值就是 true），那么当生产者向一个尚未创建的 topic 发送消息时，会自动创建一个 num.partitions（默认值为 1）个分区和 default.replication.factor（默认值为 1）个副本的对应 topic。

命名规则

topic 的命名不推荐（虽然可以这样做）使用双下划线__开头，因为以双下划线开头的 topic 一般看作是 kafka 的内部 topic，比如__consumer_offsets 和__transaction_state。

topic 的名称必须满足如下规则：

• 由大小写字母、数字、.、-、_组成

• 不能为空、不能为.、不能为..

• 长度不能超过 249

创建 topic 时除了需要 zookeeper 的地址参数外，还需要指定 topic 的名称、副本因子 replication-factor 以及分区个数 partitions 等必选参数 ，还可以包括 disable-rack-aware、config、if-not-exists 等可选参数。

检查是否创建主题成功

./kafka-list-topic.sh --zookeeper localhost:2181

或者低版本可以使用

 ./kafka-topics.sh --list --zookeeper localhost:2181

启动 produce

Kafka 的 console-producer 在 topic test1 生产消息

./bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092  --topic test1

然后输入想要产生的消息内容（如 hello world），回车：

启动 consumer

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

运行命令：

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1 --from-beginning

9092 是 kafka 单机启动的端口；--bootstrap-server 新旧 kafka 版本不一样，这个是新版本的命令

关闭 kafka 和 zookeeper

./kafka-server-stop.sh ../config/server.properties./zookeeper-server-stop.sh

心得总结

1. produce 启动的时候参数使用的是 kafka 的端口而 consumer 启动的时候使用的是 zookeeper 的端口；

2. 必须先创建 topic 才能使用；

3. topic 本质是以文件的形式储存在 zookeeper 上的。

kafka 与 zookeeper 间的关联

一个典型的 Kafka 集群中包含若干 Produce，若干 broker（一般 broker 数量越多，集群吞吐率越高），若干 Consumer Group，以及一个 Zookeeper 集群。

• Kafka 通过 Zookeeper 管理集群配置，选举 leader，以及在 Consumer Group 发生变化时进行 rebalance。

• Producer 使用 push 模式将消息发布到 broker，Consumer 使用 pull 模式从 broker 订阅并消费消息。

1. Producer 端直接连接 broker.list 列表,从列表中返回 TopicMetadataResponse,该 Metadata 包含 Topic 下每个 partition leader 建立 socket 连接并发送消息。

2. Broker 端使用 zookeeper 用来注册 broker 信息,以及监控 partition leader 存活性。

3. Consumer 端使用 zookeeper 用来注册 consumer 信息,其中包括 consumer 消费的 partition 列表等,同时也用来发现 broker 列表,并和 partition leader 建立 socket 连接,并获取消息。

Zookeeper 作用：

管理 broker、consumer

• 创建 Broker 后，向 zookeeper 注册新的 broker 信息，实现在服务器正常运行下的水平拓展。具体的，通过注册 watcher，获取 partition 的信息。

• Topic 的注册，zookeeper 会维护 topic 与 broker 的关系，通/brokers/topics/topic.name 节点来记录。

• Producer 向 zookeeper 中注册 watcher,了解 topic 的 partition 的消息，以动态了解运行情况，实现负载均衡。Zookeepr 不管理 producer，只是能够提供当前 broker 的相关信息。

• Consumer 可以使用 group 形式消费 kafka 中的数据。

• 所有的 group 将以轮询的方式消费 broker 中的数据，具体的按照启动的顺序。

• Zookeeper 会给每个 consumer group 一个 ID,即同一份数据可以被不同的消费者 ID 多次消费。

单播与多播的实现

• 以单个消费者还是以组别的方式去消费数据，由用户自己去定义。

• Zookeeper 管理 consumer 的 offset 跟踪当前消费的 offset(新版本已经放在 topic 去维护了)。