認識 kafka

kafka簡介

Kafka 是一個分布式流媒體平臺，kafka官網：http://kafka.apache.org/

1）流媒體平臺有三個關鍵功能：

•  「發布和訂閱記錄流」，類似于消息隊列或企業消息傳遞系統。
•  「以容錯的持久方式存儲」 記錄流。
•  記錄發生時處理流。

2）Kafka通常用于兩大類應用：

•  構建可在 「系統或應用程序之間」 可靠獲取數據的實時流數據管道
•  構建轉換或響應數據流的實時流應用程序

3）首先是幾個概念：

•  Kafka作為一個集群運行在一個或多個可跨多個**「數據中心的服務器」**上。
•  Kafka集群以稱為** topics主題**的類別存儲記錄流。
•  每條記錄都包含**「一個鍵，一個值和一個時間戳」**。

4）Kafka有四個核心API：

•  「Producer API（生產者API」）允許應用程序發布記錄流至一個或多個kafka的topics（主題）。
•  「Consumer API（消費者API」）允許應用程序訂閱一個或多個topics（主題），并處理所產生的對他們記錄的數據流。
•  Streams API（流API） 允許應用程序充當流處理器，從一個或多個topics（主題）消耗的輸入流，并產生一個輸出流至一個或多個輸出的topics（主題），有效地變換所述輸入流，以輸出流。
•  「Connector API（連接器API」）允許構建和運行kafka topics（主題）連接到現有的應用程序或數據系統中重用生產者或消費者。例如，關系數據庫的連接器可能捕獲對表的每個更改。

圖片

在Kafka中，客戶端和服務器之間的通信是通過簡單，高性能，語言無關的TCP協議完成的。此協議已版本化并保持與舊版本的向后兼容性。Kafka提供Java客戶端，但客戶端有多種語言版本。

1.2 Topics主題 和 partitions分區

我們首先深入了解 Kafka 為記錄流提供的核心抽象 - 主題topics

一個Topic可以認為是一類消息，每個topic將被分成多個partition(區),每個partition在存儲層面是append log文件

主題是發布記錄的類別或訂閱源名稱。Kafka的主題總是多用戶; 也就是說，一個主題可以有零個，一個或多個消費者訂閱寫入它的數據。

對于每個主題，Kafka群集都維護一個如下所示的分區日志：

圖片

每個分區都是一個有序的，不可變的記錄序列，不斷附加到結構化的提交日志中。分區中的記錄每個都分配了一個稱為偏移的順序ID號，它唯一地標識分區中的每個記錄。

Kafka集群持久保存所有已發布的記錄 - 無論是否已使用 - 使用可配置的保留期。例如，如果保留策略設置為兩天，則在發布記錄后的兩天內，它可供使用，之后將被丟棄以釋放空間。Kafka的性能在數據大小方面實際上是恒定的，因此長時間存儲數據不是問題。

圖片

實際上，基于每個消費者保留的唯一元數據是該消費者在日志中的偏移或位置。這種偏移由消費者控制：通常消費者在讀取記錄時會線性地提高其偏移量，但事實上，由于該位置由消費者控制，因此它可以按照自己喜歡的任何順序消費記錄。例如，消費者可以重置為較舊的偏移量來重新處理過去的數據，或者跳到最近的記錄并從“現在”開始消費。

這些功能組合意味著Kafka 消費者consumers 非常cheap - 他們可以來來往往對集群或其他消費者沒有太大影響。例如，您可以使用我們的命令行工具“tail”任何主題的內容，而無需更改任何現有使用者所消耗的內容。

日志中的分區有多種用途。首先，它們允許日志擴展到超出適合單個服務器的大小。每個單獨的分區必須適合托管它的服務器，但主題可能有許多分區，因此它可以處理任意數量的數據。其次，它們充當了并行性的單位 - 更多的是它。

1.3 Distribution分配

一個Topic的多個partitions,被分布在kafka集群中的多個server上;每個server(kafka實例)負責partitions中消息的讀寫操作;此外kafka還可以配置partitions需要備份的個數(replicas),每個partition將會被備份到多臺機器上,以提高可用性.

基于replicated方案,那么就意味著需要對多個備份進行調度;每個partition都有一個server為"leader";leader負責所有的讀寫操作,如果leader失效,那么將會有其他follower來接管(成為新的leader);follower只是單調的和leader跟進,同步消息即可..由此可見作為leader的server承載了全部的請求壓力,因此從集群的整體考慮,有多少個partitions就意味著有多少個"leader",kafka會將"leader"均衡的分散在每個實例上,來確保整體的性能穩定。

1.4 Producers生產者 和 Consumers消費者

1.4.1 Producers生產者

Producers 將數據發布到指定的topics 主題。同時Producer 也能決定將此消息歸屬于哪個partition;比如基于"round-robin"方式或者通過其他的一些算法等。

1.4.2 Consumers

•  本質上kafka只支持Topic.每個consumer屬于一個consumer group;反過來說,每個group中可以有多個consumer.發送到Topic的消息,只會被訂閱此Topic的每個group中的一個consumer消費。
•  如果所有使用者實例具有相同的使用者組，則記錄將有效地在使用者實例上進行負載平衡。
•  如果所有消費者實例具有不同的消費者組，則每個記錄將廣播到所有消費者進程。

圖片

分析：兩個服務器Kafka群集，托管四個分區（P0-P3），包含兩個使用者組。消費者組A有兩個消費者實例，B組有四個消費者實例。

在Kafka中實現消費consumption 的方式是通過在消費者實例上劃分日志中的分區，以便每個實例在任何時間點都是分配的“公平份額”的獨占消費者。維護組中成員資格的過程由Kafka協議動態處理。如果新實例加入該組，他們將從該組的其他成員接管一些分區; 如果實例死亡，其分區將分發給其余實例。

Kafka僅提供分區內記錄的總訂單，而不是主題中不同分區之間的記錄。對于大多數應用程序而言，按分區排序與按鍵分區數據的能力相結合就足夠了。但是，如果您需要對記錄進行總訂單，則可以使用僅包含一個分區的主題來實現，但這將意味著每個使用者組只有一個使用者進程。

1.5 Consumers kafka確保

•  發送到partitions中的消息將會按照它接收的順序追加到日志中。也就是說，如果記錄M1由與記錄M2相同的生成者發送，并且首先發送M1，則M1將具有比M2更低的偏移并且在日志中更早出現。
•  消費者實例按照它們存儲在日志中的順序查看記錄。對于消費者而言,它們消費消息的順序和日志中消息順序一致。
•  如果Topic的"replicationfactor"為N,那么允許N-1個kafka實例失效，我們將容忍最多N-1個服務器故障，而不會丟失任何提交到日志的記錄。

1.6 kafka作為消息系統

Kafka的流概念與傳統的企業郵件系統相比如何？

（1）傳統消息系統

消息傳統上有兩種模型：queuing排隊 and publish-subscribe發布 - 訂閱。在隊列中，消費者池可以從服務器讀取并且每個記錄轉到其中一個; 在發布 - 訂閱中，記錄被廣播給所有消費者。這兩種模型中的每一種都有優點和缺點。排隊的優勢在于它允許您在多個消費者實例上劃分數據處理，從而可以擴展您的處理。不幸的是，一旦一個進程讀取它已經消失的數據，隊列就不是多用戶。發布 - 訂閱允許您將數據廣播到多個進程，但由于每條消息都發送給每個訂閱者，因此無法進行擴展處理。

卡夫卡的消費者群體概念概括了這兩個概念。與隊列一樣，使用者組允許您將處理劃分為一組進程（使用者組的成員）。與發布 - 訂閱一樣，Kafka允許您向多個消費者組廣播消息。

（2）kafka 的優勢

Kafka模型的優勢在于每個主題都具有這些屬性 - 它可以擴展處理并且也是多用戶 - 不需要選擇其中一個。

與傳統的消息系統相比，Kafka具有更強的訂購保證。

傳統隊列在服務器上按順序保留記錄，如果多個消費者從隊列中消耗，則服務器按照存儲順序分發記錄。但是，雖然服務器按順序分發記錄，但是記錄是異步傳遞給消費者的，因此它們可能會在不同的消費者處出現故障。這實際上意味著在存在并行消耗的情況下丟失記錄的順序。消息傳遞系統通常通過具有“獨占消費者”概念來解決這個問題，該概念只允許一個進程從隊列中消耗，但當然這意味著處理中沒有并行性。

kafka做得更好。通過在主題中具有并行性概念 - 分區 - ，Kafka能夠在消費者流程池中提供訂購保證和負載平衡。這是通過將主題中的分區分配給使用者組中的使用者來實現的，以便每個分區僅由該組中的一個使用者使用。通過這樣做，我們確保使用者是該分區的唯一讀者并按順序使用數據。由于有許多分區，這仍然可以平衡許多消費者實例的負載。但請注意，消費者組中的消費者實例不能超過分區。

1.7 kafka作為存儲系統

•  任何允許發布與消費消息分離的消息的消息隊列實際上充當了正在進行的消息的存儲系統。Kafka的不同之處在于它是一個非常好的存儲系統。
•  寫入Kafka的數據將寫入磁盤并進行復制以實現容錯。Kafka允許生產者等待確認，以便在完全復制之前寫入不被認為是完整的，并且即使寫入的服務器失敗也保證寫入仍然存在。
•  磁盤結構Kafka很好地使用了規模 - 無論服務器上有50 KB還是50 TB的持久數據，Kafka都會執行相同的操作。
•  由于認真對待存儲并允許客戶端控制其讀取位置，您可以將Kafka視為一種專用于高性能，低延遲提交日志存儲，復制和傳播的專用分布式文件系統。

1.8 kafka用于流處理

•  僅僅讀取，寫入和存儲數據流是不夠的，目的是實現流的實時處理。
•  在Kafka中，流處理器是指從輸入主題獲取連續數據流，對此輸入執行某些處理以及生成連續數據流以輸出主題的任何內容。
•  例如，零售應用程序可能會接收銷售和發貨的輸入流，并輸出重新排序流和根據此數據計算的價格調整。
•  可以使用生產者和消費者API直接進行簡單處理。但是，對于更復雜的轉換，Kafka提供了完全集成的Streams API。這允許構建執行非平凡處理的應用程序，這些應用程序可以計算流的聚合或將流連接在一起。
•  此工具有助于解決此類應用程序面臨的難題：處理無序數據，在代碼更改時重新處理輸入，執行有狀態計算等。
•  流API構建在Kafka提供的核心原語上：它使用生產者和消費者API進行輸入，使用Kafka進行有狀態存儲，并在流處理器實例之間使用相同的組機制來實現容錯。

2、kafka使用場景

2.1 消息Messaging

Kafka可以替代更傳統的消息代理。消息代理的使用有多種原因（將處理與數據生成器分離，緩沖未處理的消息等）。與大多數消息傳遞系統相比，Kafka具有更好的吞吐量，內置分區，復制和容錯功能，這使其成為大規模消息處理應用程序的理想解決方案。

根據經驗，消息傳遞的使用通常相對較低，但可能需要較低的端到端延遲，并且通常取決于Kafka提供的強大的耐用性保證。

在這個領域，Kafka可與傳統的消息傳遞系統（如ActiveMQ或 RabbitMQ）相媲美。

2.2 網站活動跟蹤

Kafka的原始用例是能夠將用戶活動跟蹤管道重建為一組實時發布 - 訂閱源。這意味著站點活動（頁面查看，搜索或用戶可能采取的其他操作）將發布到中心主題，每個活動類型包含一個主題。這些源可用于訂購一系列用例，包括實時處理，實時監控以及加載到Hadoop或離線數據倉庫系統以進行脫機處理和報告。

活動跟蹤通常非常高，因為為每個用戶頁面視圖生成了許多活動消息。

2.3 度量Metrics

Kafka通常用于運營監控數據。這涉及從分布式應用程序聚合統計信息以生成操作數據的集中式提要。

2.4 日志聚合

許多人使用Kafka作為日志聚合解決方案的替代品。日志聚合通常從服務器收集物理日志文件，并將它們放在中央位置（可能是文件服務器或HDFS）進行處理。Kafka抽象出文件的細節，并將日志或事件數據作為消息流更清晰地抽象出來。這允許更低延遲的處理并更容易支持多個數據源和分布式數據消耗。與Scribe或Flume等以日志為中心的系統相比，Kafka提供了同樣出色的性能，由于復制而具有更強的耐用性保證，以及更低的端到端延遲。

2.5 流處理

許多Kafka用戶在處理由多個階段組成的管道時處理數據，其中原始輸入數據從Kafka主題中消費，然后聚合，豐富或以其他方式轉換為新主題以供進一步消費或后續處理。

例如，用于推薦新聞文章的處理管道可以從RSS訂閱源抓取文章內容并將其發布到“文章”主題; 進一步處理可能會對此內容進行規范化或重復數據刪除，并將已清理的文章內容發布到新主題; 最終處理階段可能會嘗試向用戶推薦此內容。此類處理管道基于各個主題創建實時數據流的圖形。從0.10.0.0開始，這是一個輕量級但功能強大的流處理庫，名為Kafka Streams 在Apache Kafka中可用于執行如上所述的此類數據處理。除了Kafka Streams之外，其他開源流處理工具包括Apache Storm和 Apache Samza。

2.6 Event Sourcing

Event Sourcing是一種應用程序設計風格，其中狀態更改記錄為按時間排序的記錄序列。Kafka對非常大的存儲日志數據的支持使其成為以這種風格構建的應用程序的出色后端。

2.7 提交日志

Kafka可以作為分布式系統的一種外部提交日志。該日志有助于在節點之間復制數據，并充當故障節點恢復其數據的重新同步機制。Kafka中的日志壓縮功能有助于支持此用法。在這種用法中，Kafka類似于Apache BookKeeper項目。

3、kafka安裝

3.1 下載安裝

到官網http://kafka.apache.org/downloads.html下載想要的版本。

注：由于Kafka控制臺腳本對于基于Unix和Windows的平臺是不同的，因此在Windows平臺上使用bin\windows\ 而不是bin/ 將腳本擴展名更改為.bat。 

[root@along ~]# wget http://mirrors.shu.edu.cn/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgz    
[root@along ~]# tar -C /data/ -xvf kafka_2.11-2.1.0.tgz    
[root@along ~]# cd /data/kafka_2.11-2.1.0/   

3.2 配置啟動zookeeper

kafka正常運行，必須配置zookeeper，否則無論是kafka集群還是客戶端的生存者和消費者都無法正常的工作的；所以需要配置啟動zookeeper服務。

（1）zookeeper需要java環境 

[root@along ~]# yum -y install java-1.8.0   

（2）這里kafka下載包已經包括zookeeper服務，所以只需修改配置文件，啟動即可。

如果需要下載指定zookeeper版本；可以單獨去zookeeper官網http://mirrors.shu.edu.cn/apache/zookeeper/下載指定版本。 

[root@along ~]# cd /data/kafka_2.11-2.1.0/    
[root@along kafka_2.11-2.1.0]# grep "^[^#]" config/zookeeper.properties    
dataDir=/tmp/zookeeper   #數據存儲目錄    
clientPort=2181   #zookeeper端口    
maxClientCnxns=0   

注：可自行添加修改zookeeper配置

3.3 配置kafka

（1）修改配置文件 

[root@along kafka_2.11-2.1.0]# grep "^[^#]" config/server.properties    
broker.id=0    
listeners=PLAINTEXT://localhost:9092    
num.network.threads=3    
num.io.threads=8    
socket.send.buffer.bytes=102400    
socket.receive.buffer.bytes=102400   
socket.request.max.bytes=104857600    
log.dirs=/tmp/kafka-logs    
num.partitions=1    
num.recovery.threads.per.data.dir=1  
offsets.topic.replication.factor=1    
transaction.state.log.replication.factor=1    
transaction.state.log.min.isr=1    
log.retention.hours=168    
log.segment.bytes=1073741824    
log.retention.check.interval.ms=300000    
zookeeper.connect=localhost:2181    
zookeeper.connection.timeout.ms=6000    
group.initial.rebalance.delay.ms=0   

注：可根據自己需求修改配置文件 

broker.id：#唯一標識ID    
listeners=PLAINTEXT://localhost:9092：#kafka服務監聽地址和端口    
log.dirs：#日志存儲目錄   
zookeeper.connect：#指定zookeeper服務   

（2）配置環境變量 

[root@along ~]# vim /etc/profile.d/kafka.sh    
export KAFKA_HOME="/data/kafka_2.11-2.1.0"    
export PATH="${KAFKA_HOME}/bin:$PATH"    
[root@along ~]# source /etc/profile.d/kafka.sh   

（3）配置服務啟動腳本 

[root@along ~]# vim /etc/init.d/kafka    
#!/bin/sh    
#    
# chkconfig: 345 99 01   
# description: Kafka    
#    
# File : Kafka    
#    
# Description: Starts and stops the Kafka server    
#    
source /etc/rc.d/init.d/functions   
KAFKA_HOME=/data/kafka_2.11-2.1.0    
KAFKA_USER=root    
export LOG_DIR=/tmp/kafka-logs    
[ -e /etc/sysconfig/kafka ] && . /etc/sysconfig/kafka  
 # See how we were called.    
case "$1" in   
   start)    
    echo -n "Starting Kafka:"    
    /sbin/runuser -s /bin/sh $KAFKA_USER -c "nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > $LOG_DIR/server.out 2> $LOG_DIR/server.err &"   
    echo " done."    
    exit 0    
    ;;     
   stop)  
     echo -n "Stopping Kafka: "    
    /sbin/runuser -s /bin/sh $KAFKA_USER  -c "ps -ef | grep kafka.Kafka | grep -v grep | awk '{print \$2}' | xargs kill \-9"   
     echo " done."    
    exit 0    
    ;;    
  hardstop)    
    echo -n "Stopping (hard) Kafka: "  
    /sbin/runuser -s /bin/sh $KAFKA_USER  -c "ps -ef | grep kafka.Kafka | grep -v grep | awk '{print \$2}' | xargs kill -9"   
     echo " done."   
    exit 0   
     ;;     
   status)   
     c_pid=`ps -ef | grep kafka.Kafka | grep -v grep | awk '{print $2}'`   
    if [ "$c_pid" = "" ] ; then    
      echo "Stopped"    
      exit 3    
    else    
      echo "Running $c_pid"    
      exit 0    
    fi    
    ;;     
   restart)    
    stop    
    start    
    ;;    
   *)    
    echo "Usage: kafka {start|stop|hardstop|status|restart}"   
     exit 1    
    ;;    
 esac   

3.4 啟動kafka服務

（1）后臺啟動zookeeper服務 

[root@along ~]# nohup zookeeper-server-start.sh /data/kafka_2.11-2.1.0/config/zookeeper.properties &   

（2）啟動kafka服務 

[root@along ~]# service kafka start    
Starting kafka (via systemctl):                            [  OK  ]    
[root@along ~]# service kafka status    
Running 86018    
[root@along ~]# ss -nutl    
Netid State      Recv-Q Send-Q     Local Address:Port                    Peer Address:Port                             
tcp   LISTEN     0      50                    :::9092                              :::*               
tcp   LISTEN     0      50                    :::2181                              :::*    

4、kafka使用簡單入門

4.1 創建主題topics

創建一個名為“along”的主題，它只包含一個分區，只有一個副本： 

[root@along ~]# kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic along    
Created topic "along".   

如果我們運行list topic命令，我們現在可以看到該主題： 

[root@along ~]# kafka-topics.sh --list --zookeeper localhost:2181    
along   

4.2 發送一些消息

Kafka附帶一個命令行客戶端，它將從文件或標準輸入中獲取輸入，并將其作為消息發送到Kafka集群。默認情況下，每行將作為單獨的消息發送。

運行生產者，然后在控制臺中鍵入一些消息以發送到服務器。 

[root@along ~]# kafka-console-producer.sh --broker-list localhost:9092 --topic along    
>This is a message    
>This is another message   

4.3 啟動消費者

Kafka還有一個命令行使用者，它會將消息轉儲到標準輸出。 

[root@along ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic along --from-beginning    
This is a message    
This is another message   

所有命令行工具都有其他選項; 運行不帶參數的命令將顯示更詳細地記錄它們的使用信息。

5、設置多代理kafka群集

到目前為止，我們一直在與一個broker運行，但這并不好玩。對于Kafka，單個代理只是一個大小為1的集群，因此除了啟動一些代理實例之外沒有太多變化。但是為了感受它，讓我們將我們的集群擴展到三個節點（仍然在我們的本地機器上）。

5.1 準備配置文件 

[root@along kafka_2.11-2.1.0]# cd /data/kafka_2.11-2.1.0/    
[root@along kafka_2.11-2.1.0]# cp config/server.properties config/server-1.properties    
[root@along kafka_2.11-2.1.0]# cp config/server.properties config/server-2.properties    
[root@along kafka_2.11-2.1.0]# vim config/server-1.properties    
    broker.id=1    
    listeners=PLAINTEXT://:9093    
    log.dirs=/tmp/kafka-logs-1    
[root@along kafka_2.11-2.1.0]# vim config/server-2.properties    
    broker.id=2    
    listeners=PLAINTEXT://:9094    
    log.dirs=/tmp/kafka-logs-2   

注：該broker.id 屬性是群集中每個節點的唯一且永久的名稱。我們必須覆蓋端口和日志目錄，因為我們在同一臺機器上運行這些，并且我們希望讓所有代理嘗試在同一端口上注冊或覆蓋彼此的數據。

5.2 開啟集群另2個kafka服務 

[root@along ~]# nohup kafka-server-start.sh /data/kafka_2.11-2.1.0/config/server-1.properties &    
[root@along ~]# nohup kafka-server-start.sh /data/kafka_2.11-2.1.0/config/server-2.properties &    
[root@along ~]# ss -nutl    
Netid State      Recv-Q Send-Q     Local Address:Port                    Peer Address:Port                          
tcp   LISTEN     0      50      ::ffff:127.0.0.1:9092                              :::*                    
tcp   LISTEN     0      50      ::ffff:127.0.0.1:9093                              :::*                                   
tcp   LISTEN     0      50      ::ffff:127.0.0.1:9094                              :::*   

5.3 在集群中進行操作

1）現在創建一個復制因子為3的新主題my-replicated-topic 

[root@along ~]# kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic    
Created topic "my-replicated-topic".   

2）在一個集群中，運行“describe topics”命令查看哪個broker正在做什么 

[root@along ~]# kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic    
Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:    
    Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 2,0,1 Isr: 2,0,1     
#注釋：第一行給出了所有分區的摘要，每個附加行提供有關一個分區的信息。由于我們只有一個分區用于此主題，因此只有一行。     
#“leader”是負責給定分區的所有讀取和寫入的節點。每個節點將成為隨機選擇的分區部分的領導者。      
#“replicas”是復制此分區日志的節點列表，無論它們是否為領導者，或者即使它們當前處于活動狀態。    
# “isr”是“同步”復制品的集合。這是副本列表的子集，該列表當前處于活躍狀態并且已經被領導者捕獲。   
#請注意，Leader: 2，在我的示例中，節點2 是該主題的唯一分區的Leader。   

3）可以在我們創建的原始主題上運行相同的命令，以查看它的位置 

[root@along ~]# kafka-topics.sh --describe --zookeeper localhost:2181 --topic along    
Topic:along PartitionCount:1    ReplicationFactor:1 Configs:    
    Topic: along    Partition: 0    Leader: 0   Replicas: 0 Isr: 0   

4）向我們的新主題發布一些消息： 

[root@along ~]# kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic    
>my test message 1    
>my test message 2    
>^C   

5）現在讓我們使用這些消息： 

[root@along ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic    
my test message 1    
my test message 2   

5.4 測試集群的容錯性

1）現在讓我們測試一下容錯性。Broker 2 充當leader 所以讓我們殺了它： 

[root@along ~]# ps aux | grep server-2.properties |awk '{print $2}'    
106737    
[root@along ~]# kill -9 106737    
[root@along ~]# ss -nutl    
tcp   LISTEN     0      50      ::ffff:127.0.0.1:9092                              :::*                       
tcp   LISTEN     0      50      ::ffff:127.0.0.1:9093                              :::*   

2）leader 已切換到其中一個從屬節點，節點2不再位于同步副本集中： 

[root@along ~]# kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic    
Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:   
    Topic: my-replicated-topic  Partition: 0    Leader: 0   Replicas: 2,0,1 Isr: 0,1   

3）即使最初接受寫入的leader 已經失敗，這些消息仍可供消費： 

[root@along ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic  
my test message 1    
my test message 2   

6、使用Kafka Connect導入/導出數據

從控制臺寫入數據并將其寫回控制臺是一個方便的起點，但有時候可能希望使用其他來源的數據或將數據從Kafka導出到其他系統。對于許多系統，您可以使用Kafka Connect導入或導出數據，而不是編寫自定義集成代碼。

Kafka Connect是Kafka附帶的工具，用于向Kafka導入和導出數據。它是一個可擴展的工具，運行連接器，實現與外部系統交互的自定義邏輯。在本快速入門中，我們將了解如何使用簡單的連接器運行Kafka Connect，這些連接器將數據從文件導入Kafka主題并將數據從Kafka主題導出到文件。

1）首先創建一些種子數據進行測試： 

[root@along ~]# echo -e "foo\nbar" > test.txt    
或者在Windows上：    
> echo foo> test.txt    
> echo bar>> test.txt  

2）接下來，啟動兩個以獨立模式運行的連接器，這意味著它們在單個本地專用進程中運行。提供三個配置文件作為參數。

第一個始終是Kafka Connect流程的配置，包含常見配置，例如要連接的Kafka代理和數據的序列化格式。

其余配置文件均指定要創建的連接器。這些文件包括唯一的連接器名稱，要實例化的連接器類以及連接器所需的任何其他配置。 

[root@along ~]# connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties    
[2019-01-16 16:16:31,884] INFO Kafka Connect standalone worker initializing ... (org.apache.kafka.connect.cli.ConnectStandalone:67)    
[2019-01-16 16:16:31,903] INFO WorkerInfo values:    
... ...    
#注：Kafka附帶的這些示例配置文件使用您之前啟動的默認本地群集配置并創建兩個連接器：第一個是源連接器，它從輸入文件讀取行并生成每個Kafka主題，第二個是宿連接器從Kafka主題讀取消息并將每個消息生成為輸出文件中的一行。   

3）驗證是否導入成功（另起終端）

在啟動過程中，您將看到許多日志消息，包括一些指示正在實例化連接器的日志消息。

① 一旦Kafka Connect進程啟動，源連接器應該開始從test.txt主題讀取行并將其生成到主題connect-test，并且接收器連接器應該開始從主題讀取消息connect-test 并將它們寫入文件test.sink.txt。我們可以通過檢查輸出文件的內容來驗證數據是否已通過整個管道傳遞： 

[root@along ~]# cat test.sink.txt    
foo    
bar   

② 請注意，數據存儲在Kafka主題中connect-test，因此我們還可以運行控制臺使用者來查看主題中的數據（或使用自定義使用者代碼來處理它）： 

[root@along ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning    
{"schema":{"type":"string","optional":false},"payload":"foo"}    
{"schema":{"type":"string","optional":false},"payload":"bar"}   

4）繼續追加數據，驗證 

[root@along ~]# echo Another line>> test.txt        
[root@along ~]# cat test.sink.txt    
foo    
bar    
Another line    
[root@along ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning    
{"schema":{"type":"string","optional":false},"payload":"foo"}    
{"schema":{"type":"string","optional":false},"payload":"bar"}    
{"schema":{"type":"string","optional":false},"payload":"Another line"}