十年河東，十年河西，莫欺少年窮
學無止境，精益求精
CSDN上資源下載地址：https://download.csdn.net/download/wolongbb/15420395 【包括kafka /java8/ ZooKeeper】並且和本博客所用版本一致。
一、安裝JAVA JDK
1、下載安裝包
http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
注意：根據32/64位操作系統下載對應的安裝包
2、添加系統變量：JAVA_HOME=C:\Java\jdk1.8.0_40
二、安裝ZooKeeper
kafka依賴ZooKeeper，因此，在啟動kafka之前，應先安裝啟動ZooKeeper，啟動ZooKeeper之前，應先配置ZooKeeper的環境變量。
1、 下載安裝包 http://zookeeper.apache.org/releases.html#download
2、 解壓並進入ZooKeeper目錄，我的目錄為：D:\tool\kafka\apache-zookeeper-3.6.2-bin
3、 在目錄中的conf文件夾中找到zoo_sample.cfg，並將“zoo_sample.cfg”重命名為“zoo.cfg”
4、 打開“zoo.cfg”找到並編輯 dataDir=./zookeeper-3.4.13/data
5、admin.serverPort端口默認為8080，如果8080被占用，需要指定admin.serverPort端口，在zoo.cfg最后一行增加：admin.serverPort=8001
6、audit.enable 默認為 false，我們設置為 true，在zoo.cfg最后一行增加：audit.enable=true
7、添加系統變量：ZOOKEEPER_HOME=D:\tool\kafka\apache-zookeeper-3.6.2-bin
8、編輯path系統變量，添加路徑：%ZOOKEEPER_HOME%\bin
9、在zoo.cfg文件中修改默認的Zookeeper端口（默認端口2181），如果端口2181未被占用，則無需修改。
10、打開新的cmd，輸入“zkServer“，運行Zookeeper，cmd需要指定目錄：D:\tool\kafka\apache-zookeeper-3.6.2-bin 【注意：我的文件夾命名中帶有bin，而不是指向bin文件夾】，也可以在D:\tool\kafka\apache-zookeeper-3.6.2-bin 文件夾中，按住 shift 鍵，右鍵鼠標，打開：windows power shell。
11、按照第10步，輸入 zkServe 后，如果出現如下界面，則證明ZooKeeper啟動成功。
注意：不要關了這個窗口，啟動kafka時會用到。
上述11個步驟中，我所遇到的坑如下：
坑1：
下載安裝包時，應下載命名中帶有bin的文件，不要去下載source文件，無論安裝ZooKeeper 還是 kafka 安裝路徑中都不要有空格。切記。要考。
通過HTTP下載即可：
坑2：
我的服務器中，admin.serverPort 的默認端口8080被占用了，導致啟動時報錯，因此，我顯式指定了端口為：8001
坑3：
建議增加這個配置：audit.enable=true，關於這個配置，我並不知道干啥用的
三、安裝Kafka
1、 下載安裝包 ：http://kafka.apache.org/downloads 【注意要下載二進制版本】
2、 解壓並進入Kafka目錄，我的路徑：D:\tool\kafka\kafka_2.12-2.7.0
3、 進入config目錄找到文件server.properties並打開
4、 找到並編輯：log.dirs=./logs
5、 找到並編輯zookeeper.connect=localhost:2181 【非集群部署的情況下，無需修改】
6、 Kafka會按照默認，在9092端口上運行，並連接zookeeper的默認端口：2181 【注意9092端口不應該被其他程序占用】
7、 進入Kafka安裝目錄D:\tool\kafka\kafka_2.12-2.7.0\bin\windows，按下Shift+右鍵，選擇“打開命令窗口”選項，打開命令行，輸入：
kafka-server-start.bat ..\..\config\server.properties
如果不行
請使用：
.\kafka-server-start.bat ..\..\config\server.properties
注意：不要關了這個窗口，啟用Kafka前請確保ZooKeeper實例已經准備好並開始運行
上述7個步驟中，我遇到的坑如下：
坑1、
環境變量Path要增加，如下截圖：
分別如下：
1、ZOOKEEPER_HOME ，kafka依賴，安裝ZooKeeper 時，就需要配置
%ZOOKEEPER_HOME%\bin
2、系統路徑，這個據說和 wbem 文件夾下的exe有關，沒有的話，加上這兩個系統路徑變量
C:\Windows\System32
C:\Windows\System32\wbem
3、據說和啟動kafka有關，經過我的測試，應該沒關系，不管有沒有關系，加上保險一點。
D:\tool\kafka\kafka_2.12-2.4.0\bin\windows
坑2
進入Kafka安裝目錄D:\tool\kafka\kafka_2.12-2.7.0\bin\windows，按下Shift+右鍵，選擇“打開命令窗口”選項，打開命令行，輸入：kafka-server-start.bat ..\..\config\server.properties
但：
好多博客上都是進入D:\tool\kafka\kafka_2.12-2.7.0 安裝路徑 ，然后執行：
.\bin\windows\kafka-server-start.bat .\config\server.properties
但，在我測試的過程中，執行 .\bin\windows\kafka-server-start.bat .\config\server.properties 無任何響應。
通用啟動方式：
通用啟動方式【統一在kafka文件夾中啟動，利用自帶的bat文件】
最后按照知乎作者的方案，通過kafka自帶的bat文件啟動相關服務：
它們都需要進入 D:\tool\kafka\kafka_2.12-2.7.0 目錄，然后再啟動相應的命令。 cd D:\kafka_2.11-2.4.0 啟動zookeeper服務，運行命令： bin\windows\zookeeper-server-start.bat config\zookeeper.properties 啟動kafka服務，運行命令： bin\windows\kafka-server-start.bat config\server.properties
四、測試
（linux直接在bin目錄下.sh，windows需要進入bin\winndows下的.bat）
1、 創建主題，進入Kafka安裝目錄D:\Kafka\kafka_2.12-0.11.0.0，按下Shift+右鍵，選擇“打開命令窗口”選項，打開命令行，輸入：
.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
注意：不要關了這個窗口
2、查看主題輸入：
.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
3、 創建生產者，進入Kafka安裝目錄D:\Kafka\kafka_2.12-0.11.0.0，按下Shift+右鍵，選擇“打開命令窗口”選項，打開命令行，輸入：
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test
注意：不要關了這個窗口
4、 創建消費者，進入Kafka安裝目錄D:\Kafka\kafka_2.12-0.11.0.0，按下Shift+右鍵，選擇“打開命令窗口”選項，打開命令行，輸入：
.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning此時，往生產者窗口寫入消息，消費者窗口也能同步的接收到消息
5、查看topic，進入Kafka安裝目錄D:\Kafka\kafka_2.12-0.11.0.0，按下Shift+右鍵，選擇“打開命令窗口”選項，打開命令行，輸入：
-
.\ bin\windows\kafka-topics.bat --describe --zookeeper localhost:2181 --topic
-
test
6、 重要（操作日志的處理）：
kafka啟動后，如果你去查看kafka所在的根目錄，或者是kafka本身的目錄，會發現已經默認生成一堆操作日志(這樣看起來真心很亂)：
而且會不斷生成不同時間戳的操作日志。剛開始不知所措，一番研究后，看了啟動的腳本內容，發現啟動的時候是會默認使用到這個log4j.properties文件中的配置，而在zoo.cfg是不會看到本身的啟動會調用到這個，還以為只有那一個日志路徑：
在這里配置一下就可以了，找到config下的log4j.properties：
zookeeper.log.dir=D:\tool\kafkaLogs
將路徑更改下即可，這樣就可以歸檔在一個文件夾下邊了，路徑根據自己喜好定義：
另外如何消除不斷生成日志的問題，就是同一天的不同時間會不停生成。
修改這里，還是在log4j.properties中：
本身都為trace，字面理解為會生成一堆跟蹤日志，將其改為INFO即可。
kafka server.properties 配置文件詳解(二)
雖然在前面一部分我們啟動了kafka集群，並通過控制台的方式實現了producer和consumer，但是我們還是了解一下kafka單個節點是的配置參數屬性，
也只有了解了這些參數的配置，才能將kafka的性能發揮到最好。
標紅部分為集群必配屬性，致於其它的參數配置等學完后再回過頭來看這些參數，你就知道是什么意思了
1.broker.id
broker.id=2 一般采用ip的后三位來用來標識是哪台kafka的broker，利於定位和排錯
2.Prot
tcp用來監控的kafka端口
listeners=PLAINTEXT://192.168.43.17:9092
3.Zookeeper.connect
#kafka連接zk的集群，多個用”,”號隔開
#zookeeper.connect=192.168.43.15:2181,192.168.43.16:2181,192.168.43.17:2181
#但是這樣寫不是很友好，kafka在連上zk后，直接在zk的根目錄創建自己需要的文件夾，這樣會導致zk的根目錄也非常亂
#同時如果有別的項目也在用zk創建1個controller目錄，kafka會產生一個controller的文件夾，這樣會導致混淆
#如果需要部署2個獨立的kafka storm環境時，只有一個zk集群時
#綜上所述，可以采用命名空間的方式解決以上問題：
zookeeper.connect=192.168.43.15:2181,192.168.43.16:2181,192.168.43.17:2181/kafka_1（推薦使用這2種方式）
zookeeper.connect=192.168.43.15:2181,192.168.43.16:2181,192.168.43.17:2181/kafka_2
4.Log.dirs(commit log)
#用來存儲日志的路徑，它上面存放了多個log segment,如果采用了磁盤陣列，路徑可以是多個
#如果采用了多個存儲路徑，1個partition只能存放在1個路徑下，因為順序的磁盤讀寫操作速度
#可以接近隨機的內存讀寫速度
log.dirs=/var/lib/kafka
5.num.recovery.threads.per.data.dir=1
#kafka配置了1個線程池,當kafka啟動時打開每個segment,啟動失敗時檢查並截取每個segment
#服務關閉時，關閉segments的句柄、管道的flush操作，
num.recovery.threads.per.data.dir=1
#注意是1個路徑默認是1個線程，如果log.dirs配置了3個路徑，那么將會有24個線程
6.auto crteat.topics.enable(配置文件中沒有)
#自動創建1個topic
#當不存在1個topic時會創建，讀的時候會創建1個topic,通過api訪問元數據時，如果不存在會創建1個topic
#但是這個創建的tpoic都是采用默認的配置，因此不建議使用
# 一般情況下我們設置成false
7.num.partitions
#設置partitions 的個數
num.partitions=1
8.log.retention.ms
#日志存放時間，依據是文件最后1次更新文件的時間，這個參數是針對broker的，按小時來算的
log.retention.hours=168
#存放1個星期
9.log.retention.bytes
#partition保留數據的大小，這個參數是針對1個partitions 的
#如果有8個partition,每個partition設置的是1GB,那最多存8GB的東西，超過了就會將老的數據刪除
log.retention.bytes=1073741824
10.log.segment.bytes
#設置segment 片斷的大小，如果存放的達到了設置的大小，這個segment 將會關閉，並且將會
#創建1個新的打開的segment，如果1個segment達到了設置的大小並且已關閉，那么這個segment是可
#以被刪除的，如果這個segment設置的小，則會頻繁的創建segment，會影響kafka的性能，如果設置的較大
#超過了log.retention.ms的保存時間，這樣會造成數據不一致或丟失的情況，因此，設置每個segment的
#大小還是要根據業務量來判斷
log.segment.bytes=1073741824
11.log.segment.ms
#另一種控制segments關閉的參數設置
log.segment.ms
12.message.max.bytes
#發送信息的最大單位值，如果超過了,會造成數據丟失
#如果要發送的數據量較大，可以調整這個參數和檢查發送數據量的
message.max.bytes=1024
最后，提供下kafka可視化工具下載地址：https://download.csdn.net/download/wolongbb/15433836
參考文獻：
https://www.cnblogs.com/xuzimian/p/10138442.html 【我的成功所在】
https://blog.csdn.net/weixin_38004638/article/details/91893910 【CSDN】
https://stackoverflow.com/questions/58754086/answer/submit 【英文站點BBS】