前言

因為公司業(yè)務需要，也因為自身知識體系需要。計劃研究一下kafka。kafka和我以往使用的rabbitmq隊列不一樣，它的吞吐量更高，但是作為消息投遞的功能及卻不同。但是，因為不熟悉，所以不敢貿然上生產。所以，我計劃打造一個日志收集的模塊，用來處理web接口調用和方法調用的日志收集。所以呢，第一步就是把環(huán)境搭建起來。
看了相關資料，消息傳輸還是最終定下來kafka，看了官網的最新版本，決定使用版本2.12 ，因為kafka是使用zookeeper做注冊中心的，所以還要下載zookeeper，版本用的是3.4.13 。

搭建

zookeeper

簡單起見，我們這里的zookeeper先使用單節(jié)點運行，有需要再升級為三節(jié)點集群。上傳文件后解壓縮：

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/ 另外，執(zhí)行了以后會出現一個zookeeper.out文件在執(zhí)行命令的目錄，注意一下就好，更多的參數我也就沒有看了，先起起來再說吧。
這里需要注意的是，配置文件里有個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的默認端口，所以我這里是跑得通的。但是醬紫，是前臺執(zhí)行，ctrl+c就退出了，還是需要使用nohup執(zhí)行才可以，也可以在腳本文件后加 -daemon 參數進行后臺執(zhí)行  

同樣的，在kafka-run-class.sh最上面也需要加入JAVA_HOME的配置，否則開機啟動腳本會找不到jdk

理論上來說，醬紫我們就已經安裝好了。但是，怎么操作怎么管理呢？我在網上查閱了一些資料，貌似沒有太多的主流的可視化管理工具，雖然有一些可以用，但存在一些不好說的bug，所以決定晚些嘗試這些可視化管理工具。坊間傳言，kafka的管理員都是用命令行來進行管理及操作的。所以，接下來我們參照官網對現有的進程做一些小小的測試，來驗證我們成功了。參考的網頁為http://kafka.apache.org/quickstart
需要注意的是，要在service.properties中配置監(jiān)聽，我的配置是

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，啟動不起來。如果不配置這個的話，就是命令行可以操作，但是遠程怎么也無法效用。

創(chuàng)建topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test //創(chuàng)建名為test的topic
bin/kafka-topics.sh --list --zookeeper localhost:2181 //查看有哪些topic

這個時候就可以看到test的名字了。

發(fā)送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test //開始發(fā)送消息
This is a message //屬于一個消息之后回車
This is another message //這是第二條消息

這里的發(fā)送消息的結果，我們去消費者那邊去看

啟動一個消費者

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning  //輸入完以后，會啟動一個消費者，就可以看到兩條消息一次顯示出來了

修改配置

這個時候如果想通過api調用創(chuàng)建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中是依賴的服務名稱，用空格隔開，可以輸入多個
注意的事項是，第一次需要先執(zhí)行enable，然后再start如下：

systemctl enable zookeeper.service
systemctl start zookeeper.service

systemctl enable kafka.service
systemctl start kafka.service

結尾

至此，我們已經搭建完成一個簡單的kafka環(huán)境了。不過，真的太簡單了，以至于覺得跟玩具似的。沒關系。我們后面會使用java進行調用，并調整相關參數。這里，僅作為單機的安裝教程存在吧。