kafka入門

一種分布式的、基于發布/訂閱的消息系統，scala編寫，具備快速、可擴展、可持久化的特點。

基本概念

topic
主題

partition
分區，一個topic下可以有多個partition，消息是分散到多個partition里存儲的，partition支持水平擴展。
一個partition內的消息是有序的，partition間的消息則是無序的。
每個partition會有若干副本。

broker
一個kafka節點

consumer
消費者，從topic里取得消息。
每個consumer維護自己的offset。
consumer數量小于分區數，會有一個消費者處理多個分區；反之，會有空閑的消費者，造成浪費。

producer
生產者，負責將消息寫入topic

特點

基于硬盤的消息保存，避免在producer上累積消息或者消息丟失。
同一個消息可以由多個consumer消費。

可擴展性：隨著數據的增加，可擴展為數十臺，上百臺規模的大集群。擴展可以在集群正常運行的時候進行，對于整個系統的運作沒有影響；這也就意味著，對于很多臺broker 的集群，如果一臺broker 有故障，不影響為client 提供服務.集群如果要同時容忍更多的故障的話, 可以配置更高的replication
factors。

高性能：上面的這些特性使得Apache Kafka 成為一個能夠在高負載的情況下表現出優越性能的發布-訂閱消息系統。Producer, consumer 和broker 都能在大數據流的情況下輕松的擴展.

kafka的版本

比如kafka_2.10-0.10.2.0
這里，2.10指的是編譯kafka的scala版本，真正的kafka版本號是后面的：0.10.2.0

配置kafka

config/server.properties文件里的配置項說明：
broker.id
每個kafka 的broker 都需要有一個整型的唯一標識，這個標識通過broker.id 來設置。默認的情況下，這個數字是0，但是它可以設置成任何值。需要注意的
是，需要保證集群中這個id 是唯一的。這個值是可以任意填寫的，并且可以在必要的時候從broker 集群中刪除。比較好的做法是使用主機名相關的標識來做
為id，比如，你的主機名當中有數字相關的信息，如hosts1.example.com，host2.example.com，那么這個數字就可以用來作為broker.id 的值。

port
默認啟動kafka 時，監聽的是TCP 的9092 端口，端口號可以被任意修改。如果端口號設置為小于1024，那么kafka 需要以root 身份啟動。但是并不推薦以root 身份啟動。

zookeeper.connect
這個參數指定了Zookeeper 所在的地址，它存儲了broker 的元信息。默認是運行在本機的2181 端口上，因此這個值被設置成
localhost:2181。這個值可以通過分號設置多個值，每個值的格式都是hostname:port/path

log.dirs
這個參數用于配置Kafka 保存數據的位置，Kafka 中所有的消息都會存在這個目錄下。可以通過逗號來指定多個目錄，kafka 會根據最少被使用的原則選擇目錄分配新的partition。注意kafka 在分配partition 的時候選擇的規則不是按照磁盤的空間大小來定的，而是分配的parition 的個數大小。

num.recovery.thread.per.data.dir
kafka 可以配置一個線程池，線程池的使用場景如下：

• 當正常啟動的時候，開啟每個parition 的文檔塊
• 當失敗后重啟時，檢查parition 的文檔塊
• 當關閉kafka 的時候，清除關閉文檔塊
  默認，每個目錄只有一個線程。最好是設置多個線程數，這樣在服務器啟動或者關閉的時候，都可以并行的進行操作。尤其是當非正常停機后，重啟時，如果有大量的分區數，那么啟動broker 將會花費大量的時間。
  【注意】這個參數是針對每個目錄的。比如，num.recovery.threads.per.data.dir 設置為8，如果有3個log.dirs 路徑，那么一共會有24 個線程。

num.partitions
這個參數用于配置新創建的topic 有多少個分區，默認是1 個。注意partition 的個數只可以被增加，不能被減少。這就意味著如果想要減少主題的分區數，那
么就需要重新創建topic。
在第一章中介紹過，kafka 通過分區來對topic 進行擴展，因此需要使用分區的個數來做負載均衡，如果新增了broker，那么就會引發重新負載分配。這并不意味著所有的主題的分區數都需要大于broker 的數量，因為kafka 是支持多個主題的，其他的主題會使用其余的broker。需要注意的是，如果消息的吞吐量很高，那么可以通過設置一個比較大的分區數，來分攤壓力。

log.retention.ms
這個參數用于配置kafka 中消息保存的時間，也可以使用log.retention.hours，默認這個參數是168 個小時，即一周。另外，還支持log.retention.minutes 和log.retention.ms。這三個參數都會控制刪除過期數據的時間，推薦還是使用log.retention.ms。如果多個同時設置，那么會選擇最小的那個。

log.retention.bytes
這個參數也是用來配置消息過期的，它會應用到每個分區，比如，你有一個主題，有8 個分區，并且設置了log.retention.bytes 為1G，那么這個主題總共可以保留8G 的數據。注意，所有的過期配置都會應用到patition 粒度，而不是主題粒度。這也意味著，如果增加了主題的分區數，那么主題所能保留的數據也就隨之增加了。
如果設置了log.retention.bytes 和log.retention.ms（或者其他過期時間的配置），只要滿足其中一個條件，消息就會被刪除。

log.segment.bytes
這個參數用來控制log 段文件的大小，而不是消息的大小。在kafka 中，所有的消息都會進入broker，然后以追加的方式追加到分區當前最新的segment 段文件中。一旦這個段文件到達log.segment.bytes 設置的大小，比如默認的1G，這個段文件就會被關閉，然后創建一個新的。一旦這個文件被關閉，就可以理解成這個文件已經過期了。這個參數設置的越小，那么關閉文件創建文件的操作就會越頻繁，這樣也會造成大量的磁盤讀寫的開銷。
通過生產者發送過來的消息的情況可以判斷這個值的大小。比如，主題每天接收100M 的消息，并且log.segment.bytes 為默認設置，那么10 天后，這個段文件才會被填滿。由于段文件在沒有關閉的時候，是不能刪除的，log.retention.ms 又是默認的設置，那么這個消息將會在17 天后，才過期刪除。因為10 天后，段文件才關閉。再過7 天，這個文件才算真正過期，才能被清除。

message.max.bytes
這個參數用于限制生產者消息的大小，默認是1000000，也就是1M。生產者在發送消息給broker 的時候，如果出錯，會嘗試重發；但是如果是因為大小的原因，那生產者是不會重發的。另外，broker上的消息可以進行壓縮，這個參數是指壓縮后的大小，這樣能多存儲很多消息。
需要注意的是，允許發送更大的消息會對性能有很大影響。更大的消息，就意味著broker 在處理網絡連接的時候需要更長的時間，它也會增加磁盤的寫操作壓力，影響IO 吞吐量。

啟動kafka

進入kafka/bin目錄。

啟動zk

nohup ./zookeeper-server-start.sh ../config/zookeeper.properties &

測試zk是否啟動

telnet localhost 2181
輸入srvr，應該會返回：

[2017-12-01 15:59:18,829] INFO Processing srvr command from /0:0:0:0:0:0:0:1:40194 (org.apache.zookeeper.server.NIOServerCnxn)
Zookeeper version: 3.4.6-1569965, built on 02/20/2014 09:09 GMT
Latency min/avg/max: 0/0/0
Received: 1
Sent: 0
Connections: 1
Outstanding: 0
Zxid: 0xdf
Mode: standalone
Node count: 127
[2017-12-01 15:59:18,833] INFO Closed socket connection for client /0:0:0:0:0:0:0:1:40194 (no session established for client) (org.apache.zookeeper.server.NIOServerCnxn)
Connection closed by foreign host.

不建議zookeeper運行在多于7個的節點上，因為集群性能會因一致性的特征而降低。

啟動kafka

nohup  ./kafka-server-start.sh -daemon ../config/server.properties

以守護進程方式執行。

topic管理命令

創建具有指定數量分區（或復制因子）的topic
./kafka-topics.sh --create --zookeeper xx.xx.xx.xx:2181 --topic test1 --replication-factor 1 --partitions 6

查看topic的元信息
./kafka-topics.sh --zookeeper localhost:2181 --describe --topic test1

查看所有topic
./kafka-topics.sh --zookeeper localhost:2181 --list

修改已有topic的分區

./kafka-topics.sh --alter --zookeeper localhost:2181 --topic test1 --partitions 4./kafka-topics.sh --zookeeper xx.xx.xx.xx:2181 --alter --topic test1 --config flush.ms=5000./kafka-topics.sh --zookeeper xx.xx.xx.xx:2181 --alter --topic test1 --config flush.messages=20000

刪除topic

./kafka-topics.sh --delete --zookeeper xx.xx.xx.xx:2181 --topic test1

但刪除topic并不是真的刪除，只是打個標記，要真正刪除，需要同時修改server.properties：

delete.topic.enable=true

并重啟kafka才能生效。

向topic發布消息

./kafka-console-producer.sh --broker-list localhost:9092 --topic test1
輸入：
msg1
msg2
按^D停止發送。

若kafka的server.properties配置了host.name，則localhost必須改成host.name的值，例如：
./kafka-console-producer.sh --broker-list xx.xx.xx.xx:9092 --topic test1

查看topic里的消息

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1 --from-beginning
輸出：
msg1
msg2
按^C停止接收。