[TOC]
<br >
*****
# **在 CentOS 7 上安装 Kafka 2.3.1 伪分布式环境**
到目前为止，我们一直在使用单个 broker，这并不好玩。对 Kafka 来说，单个 broker 只是一个大小为一的集群。 为了深入了解它，让我们把集群扩展到三个节点（仍然在一台机器上）。
> 提示：在开始本节之前，请先安装单节点环境，然后调整 JVM 启动参数。
> 编辑个人文件 `vim ~/.bash_profile`，调整 JVM 参数：
> ```
> export KAFKA_HEAP_OPTS="-Xmx512m -Xms512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=85"
> ```
> 编译 .bash\_profile，`source ~/.bash_profile`。
<br >
## **安装**
1. 首先为每个 broker 创建一个配置文件：
~~~
$ cd /opt/kafka_2.12-2.3.1
$ cp config/server.properties config/server-1.properties
$ cp config/server.properties config/server-2.properties
~~~
<br >
1. 编辑 broker 1 的配置文件 `vim config/server-1.properties`，并设置如下属性：
~~~
broker.id=1
listeners=PLAINTEXT://:9093
log.dir=/data/kafka/kafka-logs-1
~~~
> [broker.id](http://broker.id) 属性是集群中每个节点的名称，这一名称是唯一的。我们必须重写监听端口和日志目录，因为我们在同一台机器上运行这些，我们不希望所有的 broker 尝试在同一个端口注册，或者覆盖彼此的数据。
<br >
1. 编辑代理2的配置文件 `vim config/server-2.properties`，并设置如下属性：
~~~
broker.id=2
listeners=PLAINTEXT://:9094
log.dir=/data/kafka/kafka-logs-2
~~~
<br >
1. 我们已经建立 Zookeeper 和一个单节点了，现在我们只需要启动两个新的节点：
~~~
$ bin/kafka-server-start.sh config/server-1.properties &
...
[2020-01-11 21:54:39,662] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
~~~
~~~
$ bin/kafka-server-start.sh config/server-2.properties &
...
[2020-01-11 21:55:09,813] INFO [KafkaServer id=2] started (kafka.server.KafkaServer)
~~~
<br >
## **测试**
1. 现在创建一个副本为 3 的新 topic：
~~~
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
~~~
<br >
1. Good，现在我们有一个集群，但是我们怎么才能知道那些代理在做什么呢？运行 `describe topics` 命令来查看：
~~~
$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
~~~
> 第一行给出了所有分区的摘要，下面的每行都给出了一个分区的信息。因为我们只有一个分区，所以只有一行。
> * “leader”是负责给定分区所有读写操作的节点。每个节点都是随机选择的部分分区的领导者。
> * “replicas”是复制分区日志的节点列表，不管这些节点是 leader 还是仅仅活着。
> * “isr”是一组“同步” replicas，是 replicas 列表的子集，它活着并被指到 leader。
<br >
1. 我们可以在已创建的原始主题上运行相同的命令来查看它的位置：
~~~
$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test PartitionCount:1 ReplicationFactor:1 Configs:
Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
~~~
这没什么大不了，原来的主题没有副本且在服务器 0 上。我们创建集群时，这是唯一的服务器。
<br >
1. 让我们发表一些信息给我们的新 topic：
~~~
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
...
my test message 1
my test message 2
^C
~~~
<br >
1. 现在我们来消费这些消息：
~~~
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C
~~~
<br >
1. 让我们来测试一下容错性。 Broker 1 现在是 leader，让我们来杀了它（注意要切换到 root 用户）：
~~~
# ps aux | grep server-1.properties
root 12452 2.5 9.6 3707296 371960 pts/0 Sl 22:53 0:07 /opt/jdk1.8.0_231/bin/java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent...
# kill -9 12452
~~~
领导权已经切换到一个从属节点，而且节点1也不在同步副本集中了：
~~~
# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,0
~~~
2. 不过，即便原先写入消息的 leader 已经不在，这些消息仍可用于消费：
~~~
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
my test message 1
my test message 2
^C
~~~
<br >
<br >
<br >
# **参考资料**
* [Apache Kafka QuickStart](http://kafka.apache.org/quickstart)
<br >
