jdk安装
- 查看centos自带的openjdk安装包：
rpm -qa | grep openjdk
java-1.7.0-openjdk-headless-1.7.0.111-2.6.7.8.el7.x86_64
java-1.7.0-openjdk-1.7.0.111-2.6.7.8.el7.x86_64
卸载centos7的openjdk：
yum -y remove java-1.7.0-openjdk-headless-1.7.0.111-2.6.7.8.el7.x86_64下载jdk1.8
http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html-
解压jdk安装包
tar -xvf jdk-8u201-linux-x64.tar.gz vi /etc/profileJAVA_HOME=/home/xxx/software/jdk1.8.0_201
JRE_HOME=/home/xxx/software/jdk1.8.0_201/jre
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
export JAVA_HOME JRE_HOME PATH CLASSPATH
是profile文件生效
sourece /etc/profile测试Java是否安装成功
java -version
配置免密登录
进入到.ssh文件夹中，如果没有就自己创建该文件夹，生成公钥私钥
ssh-keygen -t rsa 对于提示 一路回车将生成的公钥 id_rsa.pub 追加到 authorized_keys文件中，
cat id_rsa.pub >> authorized_keys
这里需要注意修改authorized_keys文件的权限：
chmod 600 authorized_keys配置完成后本地测试一下，如果没有提示输入密码则配置成功
ssh localhost对于分布式环境，只需要把对应的公钥放到目标机器的authorized_keys里面即可免密登录
我们要注意，.ssh目录的权限为700，其下文件authorized_keys和私钥的权限为600。否则会因为权限问题导致无法免密码登录。我们可以看到登陆后会有known_hosts文件生成。
Zookeeper安装
下载zookeeper： https://mirrors.cnnic.cn/apache/zookeeper/
解压： tar xvf zookeeper-3.4.13.tar.gz
-
修改配置文件：
dataDir=/opt/data/zookeeper/data
存储快照文件的目录，默认情况下， 事务日志也会存储在该目
dataDir录上。由于事务日志 的写性能直接影响 ZooKeeper 性能，因此 建议同时配置参数 dataLogDir
dataLogDir=/opt/data/zookeeper/logs集群配直。首先在 3 台机器的/etc/hosts 文件中加入 3 台机器 的 IP 与机器域名映射， 域名自定义， 这里分别命名为 server-I、 server-2、 server-3, 3 台机器 IP与机器域名映射关 系如下:
10.211.55.4 server-1
10.211.55.5 server-2
10.211.55.6 server-3- 进入其中一台机器的Zookeeper安装路径conf，添加
server.1=server-1:2888:3888
server.2=server-2:2888:3888
server.3=server-3:2888:3888端口号2888表示该服务器与集群中leader交换信息的端口，默认为2888， 3888表示选举时服务器相互通信的端口。
接着在${dataDir}路径下创建一个myid文件，myid存放的值就是服务器的编号，即对应上面的1、2、3。ZooKeeper在启动时会读取 myid文件 中的值与 zoo.cfg文件中的配置信息进行比较， 以确定是哪台服务器。
将配置好的zoo.cfg拷贝到其他两台机器，并分别创建对应的myid，为了操作方便，我们可以将Zookeeper相关环境变量添加到/etc/profile文件中，如：
export ZOOKEEPER_HOME=/home/xxx/software/zookeeper-3.4.13
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin:$ZOOKEEPER_HOME/bin-
验证。 zkServer.sh start
zkServer.sh status
输出下面类似结果表示安装成功``` ZooKeeper JMX enabled by default Using config: /home/xxx/software/zookeeper-3.4.13/bin/../conf/zoo.cfg
Mode: follower
```
注意： centos7默认使用firewall作为防火墙，并默认开启， 在启动zk时需要关闭防护墙，不然无法通信，查看防火墙状态： firewall-cmd --state 停止防火墙： systemctl stop firewalld.service 禁止防火墙开机启动：systemctl disable firewalld.service
kafka安装
下载 http://kafka.apache.org/downloads
当前 Kafka 最新版本为 kafka_2.12-2.1.0.tgz，其中 2.12 代表 Scala 版本， 2.1.0 表示 Kafka 的版本解压安装 tar xvf kafka_2.12-2.1.0.tgz， 为了操作方便，这里我们对kafka的环境变量进行设置， 在/etc/profile 文件中加入kafka的安装路径，
export KAFKA_HOME=/home/xxx/software/kafka_2.12-2.1.0
PATH=\$PATH:\$JAVA_HOME/bin:\$JRE_HOME/bin:\$ZOOKEEPER_HOME/bin:\$KAFKA_HOME/bin
- 修改配置 。修改$KAFKA_HOME/config 目录下的 server.prope叫es文件，为了便于后 续集群环境搭建的配置， 需要保证同一个集群下 broker.id要唯一，因此这里手动配置 broker.id, 直接保持与ZooKeeper的myid值一致， 同时配置日志存储路径。server.properties修改的配置 如下 :
broker.id=l #指定代理的 id
log.dirs=/opt/data/kafka- logs #指定 Log 存储路径
zookeeper . connect=server-1:2181 , server - 2:2181 , server-3:2181
在三台机器上分别修改配置文件server.properties, 并修改对应的broker.id.
启动。
kafka-server-start.sh -daemon ../config/server.properties
执行 jps命令查看 Java进程，此时进程信息至少包括以下几项:
15976 Jps
14999 QuorumPeerMain
15906 Kafka-
通过 ZooKeeper 客户端登录 ZooKeeper 查看目录结构，执行以下命令:
zkCli.sh -server server 1:2181 #登录 ZooKeeper
ls / #查看 ZooKeeper 目录结构
ls /brokers/ ids 输出 [1, 2, 3]由/brokers/ids 节点存储的元数据可知， 3台机器的 Kafka 均已正常己启动。至此， Kafka 分布式环境搭建过程介绍完毕。
kafka实战
- 创建一个拥有3个副本的topic:
kafka-topics.sh --create --zookeeper 10.211.55.6:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
kafka-topics.sh --list --zookeeper 10.211.55.5:2181
my-replicated-topic
现在我们搭建了一个集群，怎么知道每个节点的信息呢？运行“"describe topics”命令就可以了：
kafka-topics.sh --describe --zookeeper server-1:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1
第一行是对所有分区的一个描述，然后每个分区都会对应一行，因为我们只有一个分区所以下面就只加了一行。
leader：负责处理消息的读和写，leader是从所有节点中随机选择的.
replicas：列出了所有的副本节点，不管节点是否在服务中.
isr：是正在服务中的节点.
在我们的例子中，节点2是作为leader运行。
修改topic过期时间
默认是7天 修改为3天
kafka-topics.sh --zookeeper server-1:2181 -topic xxxx --alter --config retention.ms=259200000
往topic发送消息：
kafka-console-producer.sh --broker-list server-1:9092 --topic my-replicated-topic消费这些消息：
kafka-console-consumer.sh --bootstrap-server server-1:9092 --from-beginning --topic my-replicated-topic
- 测试一下容错能力：
broker2作为leader运行，现在kill掉：
jps
22839 QuorumPeerMain
22334 Jps
1999 Kafka
[xxx@centos-linux-2 config]$ kill -9 1999
kafka-topics.sh --describe --zookeeper server-1:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 3 Replicas: 2,3,1 Isr: 3,1
现在leader为3，
虽然最初负责续写消息的leader down掉了，但之前的消息还是可以消费的：
kafka-console-consumer.sh --bootstrap-server server-1:9092 --from-beginning --topic my-replicated-topic
hello
world
nihao
yes
hello
