前言

因为公司业务需要，也因为自身知识体系需要。计划研究一下kafka。kafka和我以往使用的rabbitmq队列不一样，它的吞吐量更高，但是作为消息投递的功能及却不同。但是，因为不熟悉，所以不敢贸然上生产。所以，我计划打造一个日志收集的模块，用来处理web接口调用和方法调用的日志收集。所以呢，第一步就是把环境搭建起来。
看了相关资料，消息传输还是最终定下来kafka，看了官网的最新版本，决定使用版本2.12 ，因为kafka是使用zookeeper做注册中心的，所以还要下载zookeeper，版本用的是3.4.13 。

搭建

zookeeper

简单起见，我们这里的zookeeper先使用单节点运行，有需要再升级为三节点集群。上传文件后解压缩：

cp zookeeper-3.4.13.tar.gz /opt/
cd /opt/
tar zxvf zookeeper-3.4.13.tar.gz

cd /opt/zookeeper-3.4.13/conf
cp zoo_sample.cfg zoo.cfg

cd /opt/zookeeper-3.4.13/bin
./zkServer.sh start

需要注意的是需要预先安装jdk，安装教程可以参考这篇博客：https://blog.naturetrible.com/index.php/257/ntbrick/nature/ 另外，执行了以后会出现一个zookeeper.out文件在执行命令的目录，注意一下就好，更多的参数我也就没有看了，先起起来再说吧。
这里需要注意的是，配置文件里有个dataDir，默认是在tmp下面的，我把它改到了zookeeper的安装目录里面。
还有就是在zkEnv.sh中最上面加入JAVA_HOME，这样我们才能做开机启动，否则手动可以启动，开机启动会报找不到java

安装kafka

tar zxvf kafka_2.12-2.1.0.tgz
mv kafka_2.12-2.1.0 /opt/
cd /opt/kafka_2.12-2.1.0
bin/kafka-server-start.sh config/server.properties //因为这个配置文件的zookeeper的默认配置就是本地zookeeper的默认端口，所以我这里是跑得通的。但是酱紫，是前台执行，ctrl+c就退出了，还是需要使用nohup执行才可以，也可以在脚本文件后加 -daemon 参数进行后台执行  

同样的，在kafka-run-class.sh最上面也需要加入JAVA_HOME的配置，否则开机启动脚本会找不到jdk

理论上来说，酱紫我们就已经安装好了。但是，怎么操作怎么管理呢？我在网上查阅了一些资料，貌似没有太多的主流的可视化管理工具，虽然有一些可以用，但存在一些不好说的bug，所以决定晚些尝试这些可视化管理工具。坊间传言，kafka的管理员都是用命令行来进行管理及操作的。所以，接下来我们参照官网对现有的进程做一些小小的测试，来验证我们成功了。参考的网页为http://kafka.apache.org/quickstart
需要注意的是，要在service.properties中配置监听，我的配置是

listeners=PLAINTEXT://192.168.137.92:9092
log.dirs=/opt/kafka/kafka-logs
zookeeper.connect=192.168.1.40:2181

这里好像必须是一个确切的ip，我试过配置0.0.0.0，启动不起来。如果不配置这个的话，就是命令行可以操作，但是远程怎么也无法效用。

创建topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test //创建名为test的topic
bin/kafka-topics.sh --list --zookeeper localhost:2181 //查看有哪些topic

这个时候就可以看到test的名字了。

发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test //开始发送消息
This is a message //属于一个消息之后回车
This is another message //这是第二条消息

这里的发送消息的结果，我们去消费者那边去看

启动一个消费者

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning  //输入完以后，会启动一个消费者，就可以看到两条消息一次显示出来了

修改配置

这个时候如果想通过api调用创建topic是不行的，因为配置不对，我们需要在server.properties 添加如下配置：

listeners=PLAINTEXT://192.168.137.92:9092 //这是我的虚拟机地址，大家根据自己的修改

详细的可以在文件中搜索listeners，看相关的说明PLAINTEXT是listener的名称，干什么用的，我现在还不清楚

开机启动脚本

zookeeper启动脚本(/lib/systemd/system/zookeeper.service)：

[Unit]
Description=zookeeper2181
After=network.target

[Service]

Type=forking
ExecStart=/opt/zookeeper-3.4.13/bin/zkServer.sh start
ExecStop=/opt/zookeeper-3.4.13/bin/zkServer.sh stop
PrivateTmp=true

[Install]
WantedBy=multi-user.target

kafka启动脚本（/lib/systemd/system/kafka.service）

[Unit]
Description=kafka2.12
After=network.target zookeeper.service

[Service]
Type=forking
ExecStart= /opt/kafka_2.12-2.1.0/bin/kafka-server-start.sh -daemon /opt/kafka_2.12-2.1.0/config/server.properties
ExecStop=/opt/kafka_2.12-2.1.0/bin/kafka-server-stop.sh /opt/kafka_2.12-2.1.0/config/server.properties
PrivateTmp=true

[Install]
WantedBy=multi-user.target

注意kafka脚本的after中，加入了上面的zookeeper.service
启动脚本的after中是依赖的服务名称，用空格隔开，可以输入多个
注意的事项是，第一次需要先执行enable，然后再start如下：

systemctl enable zookeeper.service
systemctl start zookeeper.service

systemctl enable kafka.service
systemctl start kafka.service

结尾

至此，我们已经搭建完成一个简单的kafka环境了。不过，真的太简单了，以至于觉得跟玩具似的。没关系。我们后面会使用java进行调用，并调整相关参数。这里，仅作为单机的安装教程存在吧。