Kafka 环境部署

前期准备

Kafka概述

访问 Kafka 官网了解 Kafka 是什么、能做什么、如何使用。

Kafka 安装前置条件

安装 ZooKeeper

下载 Kafka 与解压

下载地址

找到 kafka_2.11-0.9.0.0.tgz 并下载

将他传输到 /abs/software 目录

解压到指定目录

1
2
3

#将Kafka解压到/abs/app/中

tar -zxvf kafka_2.11-0.9.0.0.tgz -C /abs/app/

配置 kafka 环境变量

将 kafka 目录添加到系统环境变量(~/.bash_profile)中

vi ~/.bash_profile

export KAFKA_HOME=/abs/app/kafka_2.11-0.9.0.0
export PATH=$KAFKA_HOME/bin:$PATH

让配置生效

1	source ~/.bash_profile

Kafka单节点单broker的部署及使用

修改配置文件

在 /abs/app/tmp 目录下新建 kafka-logs 文件夹:

1	mkdir kafka-logs

在 /abs/app/kafka_2.11-0.9.0.0/config 目录中对 server.properties 进行修改

# 编辑配置文件
vi server.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

# The port the socket server listens on
#port=9092
listeners=PLAINTEXT://:9092

# Hostname the broker will bind to. If not set, the server will bind to all interfaces
host.name=hadoop

# A comma seperated list of directories under which to store log files
log.dirs=/abs/app/tmp/kafka-logs

# The default number of log partitions per topic. More partitions allow greater
num.partitions=1

# root directory for all kafka znodes.
zookeeper.connect=hadoop:2181

启动 Kafka

先启动 ZooKeeper

参照 Kafka 官网快速启动

在前台启动单节点单 broker 的 Kafka

在 /abs/app/kafka_2.11-0.9.0.0/bin 目录中进行如下操作:

1	kafka-server-start.sh $KAFKA_HOME/config/server.properties

验证启动 Kafka

用 jps 查看当前所有的 Java 进程 pid如下:

[root@hadoop tmp]# jps
2448 QuorumPeerMain
2595 Jps
2538 Kafka

[root@hadoop tmp]# jps -m
2448 QuorumPeerMain /abs/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg
2538 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server.properties
2606 Jps -m

可以看到 Kafka 进程已经启动

创建一个 topic

1	kafka-topics.sh --create --zookeeper hadoop:2181 --replication-factor 1 --partitions 1 --topic hello_topic

(PS: 创建 topic 相当于指定 zookeeper)

查看所有 topic

1	kafka-topics.sh --list --zookeeper hadoop:2181

发送消息

1	kafka-console-producer.sh --broker-list hadoop:9092 --topic hello_topic

(PS: 发送消息相当于指定 broker)

消费消息

1	kafka-console-consumer.sh --zookeeper hadoop:2181 --topic hello_topic --from-beginning

(PS: 消费消息相当于指定 zookeeper 。 from-beginning 的意思为是否从开始消费，根据实际需求决定是否添加)

验证发送消息与消费消息

在发送端发送消息，看看在消费消息端是否能看到。

查看 topic 详情

查看所有 topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181

查看指定 topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181 --topic hello_topic

Kafka单节点多broker的部署及使用

修改配置文件

参照 Kafka 官网快速启动进行相关配置

在 /abs/app/kafka_2.11-0.9.0.0/config 目录中为 broker 新建三个配置文件

# make a config file for each of the brokers
cp server.properties server-1.properties
cp server.properties server-2.properties
cp server.properties server-3.properties

对 server-1.properties 、 server-2.properties 、 server-3.properties 进行相关配置

# config server-1.properties
vi server-1.properties

broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/abs/app/tmp/kafka-logs-1

# config server-2.properties
vi server-2.properties

broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/abs/app/tmp/kafka-logs-2

# config server-3.properties
vi server-3.properties

broker.id=3
listeners=PLAINTEXT://:9095
log.dir=/tmp/kafka-logs-3

启动 Kafka

先启动 ZooKeeper

参照 Kafka 官网快速启动

在后台启动单节点多 broker 的 Kafka

在 /abs/app/kafka_2.11-0.9.0.0/bin 目录中进行如下操作:

kafka-server-start.sh -daemon $KAFKA_HOME/config/server-1.properties &

kafka-server-start.sh -daemon $KAFKA_HOME/config/server-2.properties &

kafka-server-start.sh -daemon $KAFKA_HOME/config/server-3.properties &

验证启动 Kafka

用 jps 查看当前所有的 Java 进程 pid如下:

[root@hadoop config]# jps
2577 Kafka
2451 QuorumPeerMain
2520 Kafka
2633 Kafka
2686 Jps
[3]+  Done                    kafka-server-start.sh -daemon $KAFKA_HOME/config/server-3.properties

[root@hadoop config]# jps -m
2577 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-2.properties
2451 QuorumPeerMain /abs/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg
2520 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-1.properties
2633 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-3.properties
2697 Jps -m

可以看到 Kafka 进程已经启动。

创建一个 topic

1	kafka-topics.sh --create --zookeeper hadoop:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

(PS: 创建 topic 相当于指定 zookeeper，这里是一个分区三个副本)

查看所有 topic

1	kafka-topics.sh --list --zookeeper hadoop:2181

发送消息

1	kafka-console-producer.sh --broker-list hadoop:9093,hadoop:9094,hadoop:9095 --topic my-replicated-topic

(PS: 发送消息相当于指定 broker)

消费消息

1	kafka-console-consumer.sh --zookeeper hadoop:2181 --topic my-replicated-topic

(PS: 消费消息相当于指定 zookeeper 。 from-beginning 的意思为是否从开始消费，根据实际需求决定是否添加)

验证发送消息与消费消息

在发送端发送消息，看看在消费消息端是否能看到。

查看 topic 详情

查看所有 topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181

查看指定 topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181 --topic my-replicated-topic

Kafka 容错性测试

前期准备

容错测试基于 Kafka单节点多broker的部署及使用

查看指定 topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181 --topic my-replicated-topic

结果如下:

1 2	Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs: Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3

可以得出有一个副本、三个分区，主节点是1号分区，在运行着的有1、2、3

用 jps -m 查看当前所有的 Java 进程 pid

2640 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-3.properties
2738 ConsoleProducer --broker-list hadoop:9093,hadoop:9094,hadoop:9095 --topic my-replicated-topic
2581 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-2.properties
2759 ConsoleConsumer --zookeeper hadoop:2181 --topic my-replicated-topic
2522 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-1.properties
2492 QuorumPeerMain /abs/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg
2958 Jps -m

验证发送消息与消费消息1

关闭从节点的进程 server-2.properties ，进程号可以看出是 2581，命令如下:

1	kill -9 2581

再查看当前所有的 Java 进程

[root@hadoop tmp]# jps -m
2640 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-3.properties
2738 ConsoleProducer --broker-list hadoop:9093,hadoop:9094,hadoop:9095 --topic my-replicated-topic
2759 ConsoleConsumer --zookeeper hadoop:2181 --topic my-replicated-topic
2522 Kafka /abs/app/kafka_2.11-0.9.0.0/config/server-1.properties
2492 QuorumPeerMain /abs/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg
2990 Jps -m

在发送端发送消息，看看在消费消息端是否能看到。

经过测试发现可以从消费信息端接收到发送消息端发出的消息。

然而查看 my-replicated-topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181 --topic my-replicated-topic

结果如下:

1 2	Topic:my-replicated-topic PartitionCount:1 Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs: Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,3

可以得出有一个副本、三个分区，主节点是1号分区，在运行着的有1、3

2号从节点的故障并没有影响 Kafka 的正常运行

验证发送消息与消费消息2

关闭主节点的进程 server-1.properties ，进程号可以看出是 2522，命令如下:

1	kill -9 2522

在发送端发送消息，看看在消费消息端是否能看到。

经过测试发现可以从消费信息端接收到发送消息端发出的消息。

然而查看 my-replicated-topic 详情

1	kafka-topics.sh --describe --zookeeper hadoop:2181 --topic my-replicated-topic

结果如下:

1 2	Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs: Topic: my-replicated-topic Partition: 0 Leader: 3 Replicas: 1,2,3 Isr: 3

可以得出有一个副本、三个分区，主节点是3号，在运行着的只有3号节点

1、2号从节点的故障并没有影响 Kafka 的正常运行，自动选举3号节点为主节点

Kafka 的容错性得到了证明