新入手mac，當然是裝各種軟件啦
下面來記錄一下使用mac安裝kafka的過程，mac上面的homebrew是真的好用
下面我們來開始進入安裝吧
安裝環(huán)境基礎 # jdk1.8 并且配置好環(huán)境變量
1.直接使用brew安裝，安裝過程會自動安裝zookeeper

brew install kafka

效果如下

2. 安裝位置以及配置文件路徑都在圖片上顯示了，并且怎么啟動也都告訴了我們
3. 安裝位置

/usr/local/Cellar/zookeeper /usr/local/Cellar/kafka

4.配置文件位置

/usr/local/etc/kafka/server.properties /usr/local/etc/kafka/zookeeper.properties

5.現(xiàn)在沒有出現(xiàn)報錯基本已經(jīng)成功百分之90了，下面就是進行測試了
6. 打開一個終端

#創(chuàng)建topic kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test #查看創(chuàng)建的topic kafka-topics --list --zookeeper localhost:2181 #終端1發(fā)送消息 kafka-console-producer --broker-list localhost:9092 --topic test

新開一個終端

#終端2接受消息 kafka-console-consumer --bootstrap-server localhost:9092 --topic test --from-beginning

此時，在終端1中輸入信息，終端2會立即顯示在屏幕上面，如有結(jié)果輸出，即安裝成功，更高級功能請參考官網(wǎng)
7.擴展

#指定分區(qū)消費消息 kafka-console-consumer --bootstrap-server localhost:9092 --topic test --partition 1 --from-beginning#添加分區(qū) kafka-topics --alter --zookeeper localhost:2181 --partitions 10 --topic test //添加10個#啟動kafka，指定配置文件，后臺啟動并打印日志到 /usr/local/etc/kafka/kafka.lognohup kafka-server-start /usr/local/etc/kafka/server.properties > /usr/local/etc/kafka/kafka.log 2>&1 &

8.資料小擴展，本段為本人學習kafka時學到的小知識，不是很深但是很實用，經(jīng)驗不多，小弟也是剛?cè)腴T，先分享為快

8.1 topic

# topic： 1.kafka集群會將每個topic進行分區(qū)，每個分區(qū)都是一個排序且不可改變的隊列，新的消息會進入隊尾并分配一個唯一ID，官方稱之為偏移量(offset) 2.無論消息是否被消費，集群都會保留消息,有一個配置的時間（過期時間）,超過這個時間后，消息會被清除 3.消費端唯一記錄的元信息就是自己在topic中的位置(offset)， 4.分布式的原因：第一集群可以容納大量的數(shù)據(jù) 第二：可以并行的處理

8.2 消費語義的理解

at last once：至少消費一次（對一條消息有可能多次消費，有可能會造成重復消費數(shù)據(jù)）原因：Proudcer產(chǎn)生數(shù)據(jù)的時候，已經(jīng)寫入在broker中，但是由于broker的網(wǎng)絡異常，沒有返回ACK，這時Producer,認為數(shù)據(jù)沒有寫入成功，此時producer會再次寫入，相當于一條數(shù)據(jù)，被寫入了多次。at most once：最多消費一次，對于消息，有可能消費一次，有可能一次也消費不了原因：producer在產(chǎn)生數(shù)據(jù)的時候，有可能寫數(shù)據(jù)的時候不成功，此時broker就跳過這個消息，那么這條數(shù)據(jù)就會丟失，導致consumer無法消費。exactly once：有且僅有一次。這種情況是我們所需要的，也就是精準消費一次。kafka中消費語義的場景at last once：可以先讀取數(shù)據(jù)，處理數(shù)據(jù)，最后記錄offset，當然如果在記錄offset之前就crash,新的consumer會重復的來消費這條數(shù)據(jù)，導致了”最少一次“at most once：可以先讀取數(shù)據(jù)，然后記錄offset，最后在處理數(shù)據(jù)，這個方式，就有可能在offset后，還沒有及時的處理數(shù)據(jù)，就crash了，導致了新的consumer繼續(xù)從這個offset處理，那么剛剛還沒來得及處理的數(shù)據(jù)，就永遠不會被處理，導致了”最多消費一次“exactly once：可以通過將提交分成兩個階段來解決：保存了offset后提交一次，消息處理成功后，再提交一次。

8.3 消息一致性

kafka中如何實現(xiàn)精準寫入數(shù)據(jù)？ A：Producer 端寫入數(shù)據(jù)的時候保證冪等性操作： 冪等性：對于同一個數(shù)據(jù)無論操作多少次都只寫入一條數(shù)據(jù)，如果重復寫入，則執(zhí)行不成功 B：broker寫入數(shù)據(jù)的時候，保證原子性操作, 要么寫入成功，要么寫入失敗。（不成功不斷進行重試）

8.4 AckMode
先放上源碼

/*** The offset commit behavior enumeration.*/public enum AckMode {/*** Commit after each record is processed by the listener.*/RECORD,/*** Commit whatever has already been processed before the next poll.*/BATCH,/*** Commit pending updates after* {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.*/TIME,/*** Commit pending updates after* {@link ContainerProperties#setAckCount(int) ackCount} has been* exceeded.*/COUNT,/*** Commit pending updates after* {@link ContainerProperties#setAckCount(int) ackCount} has been* exceeded or after {@link ContainerProperties#setAckTime(long)* ackTime} has elapsed.*/COUNT_TIME,/*** User takes responsibility for acks using an* {@link AcknowledgingMessageListener}.*/MANUAL,/*** User takes responsibility for acks using an* {@link AcknowledgingMessageListener}. The consumer is woken to* immediately process the commit.*/MANUAL_IMMEDIATE,}

源碼解讀

RECORD 每處理一條commit一次 BATCH(默認) 每次poll的時候批量提交一次，頻率取決于每次poll的調(diào)用頻率 TIME 每次間隔ackTime的時間去commit(跟auto commit interval有什么區(qū)別呢？) COUNT 累積達到ackCount次的ack去commit COUNT_TIME ackTime或ackCount哪個條件先滿足，就commit MANUAL listener負責ack，但是背后也是批量上去 MANUAL_IMMEDIATE listner負責ack，每調(diào)用一次，就立即commit

9.終結(jié)
以上就是我安裝kafka時順帶學習的一點東西，并沒有深入，所以總結(jié)的有點亂，看到什么就是記錄的什么，更深層次的kafka知識請參閱大牛的文章或者官網(wǎng)閱讀，敬請期待本人下次更新kafka相關知識！

總結(jié)

以上是生活随笔為你收集整理的mac 搭建kafka系列教程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。