一 安装 jdk
1 安装 jdk ，这里选择 openjdk 1.8yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
安装后的jre和jdk在 /usr/lib/jvm/ 里，本例中具体的文件夹是java-1.8.0-openjdk-1.8.0.144-0.b01.el7_4.x86_64，注意这里的版本可能和你安装的不一样，以你的为准
2 配置 jdk 环境变量在 /etc/profile文件 或者 /etc/profile.d/文件夹中新建 java.sh 文件，这里使用新建文件的方式，在文件中加入以下内容vi /etc/profile.d/java.sh
JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.144-0.b01.el7_4.x86_64 #注意这里的版本号按你安装的来JRE_HOME=$JAVA_HOME/jreCLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jarPATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/binexport JAVA_HOME JRE_HOME CLASSPATH PATH
3 让环境变量立即生效
source /etc/profile
4 检验版本
java -version
二 配置防火墙
1 配置防火墙firewall-cmd --permanent --zone=public --add-port=2181/tcp # 客户端访问端口firewall-cmd --permanent --zone=public --add-port=2888/tcp # zookeeper 之间访问的端口firewall-cmd --permanent --zone=public --add-port=3888/tcp # zookeeper 和其它应用的通信端口firewall-cmd --permanent --zone=public --add-port=9092/tcp # 客户端访问端口firewall-cmd --reload
2 降低SELinux的安全级别，否则SELinux会限制集群数据传输（setenforce 0 命令只能设置运行时的安全级别，想要完全设置安全级别，需要在SELinux的配置文件中进行配置更改）
打开/etc/selinux/config配置文件，设置SELINUX=permissive 使用reboot命令重启服务器使SELinux安全级别更改生效三 下载并启动 kafka
1 创建 kafka 文件夹mkdir /usr/local/kafka
2 进入创建的 kafka 文件夹，下载 kafka 二进制代码
cd /usr/local/kafkacurl -o kafka_2.11-0.11.0.1.tgz http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.11.0.1/kafka_2.11-0.11.0.1.tgz
3 解压 kafka 并进入文件夹
tar -xzf kafka_2.11-0.11.0.1.tgzcd kafka_2.11-0.11.0.1
4 启动 kafka
kafka 需要 zookeeper 作为基本服务，kafka 下载的代码中有自带的单台 zookeeper 服务器，使用解压文件夹根目录下的配置启动 zookeeper ，在参数中使用 -daemon 让服务在后台运行，不占用命令界面。（网上有说在命令最后便加 & 号也可以起到 -daemon 参数的作用，但我使用未成功，网上命令 bin/kafka-server-start.sh config/server-1.properties &）bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
5 启动kafka
bin/kafka-server-start.sh -daemon config/server.properties
四 验证 kafka（需要在加压的文件夹根目录中执行以下命令，本例中是 cd /usr/local/kafka/kafka_2.11-0.11.0.1）
1 创建一个 topic ，就像一个数据库表一样，生产者和消费者都需要通过一个 topic 向消息队列发送消息或者消费消息bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
2 查看刚才的 topic 是否创建成功，如果成功会输出 test
bin/kafka-topics.sh --list --zookeeper localhost:2181
3 模拟生产者向 test 发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
运行上边的命令后就可以在 > 号提示下发送信息了，没发送一条后回车确认
4 再启动一个远程连接 ssh 模拟消费者客户端bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
执行以上命令后刚才发的消息都能收到，而且现在在刚才的生产者 ssh 客户端发送信息，消费者客户端都可以收到。
五 设置开机自启动服务
1 编写 zookeeper 的 systemd 自启动服务文件vi /etc/systemd/system/zookeeper.service
内容如下
[Unit]Description=zookeeperAfter=syslog.target network.target[Service]Type=forkingExecStart=/usr/local/kafka/kafka_2.11-0.11.0.1/bin/zookeeper-server-start.sh -daemon /usr/local/kafka/kafka_2.11-0.11.0.1/config/zookeeper.propertiesExecStop=/usr/local/kafka/kafka_2.11-0.11.0.1/bin/zookeeper-server-stop.shRestart=alwaysUser=rootGroup=root[Install]WantedBy=multi-user.target
2 设置 zookeeper 自启动
systemctl enable zookeepersystemctl start zookeeper
3 编写 kafka 的 systemd 自启动服务文件
[Unit]Description=kafkaAfter=network.target remote-fs.target nss-lookup.target zookeeper.serviceWants=zookeeper.service[Service]Type=forkingExecStart=/usr/local/kafka/kafka_2.11-0.11.0.1/bin/kafka-server-start.sh -daemon /usr/local/kafka/kafka_2.11-0.11.0.1/config/server.propertiesExecStop=/usr/local/kafka/kafka_2.11-0.11.0.1/bin/kafka-server-stop.shRestart=alwaysUser=rootGroup=root[Install]WantedBy=multi-user.target
4 设置 kafka 自启动
systemctl enable kafkasystemctl start kafka