一、kafka介紹
1.1 主要功能
根據官網的介紹，ApacheKafka®是一個分布式流媒體平台，它主要有3種功能：
1：It lets you publish and subscribe to streams of records.發布和訂閱消息流，這個功能類似於消息隊列，這也是kafka歸類為消息隊列框架的原因
2：It lets you store streams of records in a fault-tolerant way.以容錯的方式記錄消息流，kafka以文件的方式來存儲消息流
3：It lets you process streams of records as they occur.可以再消息發布的時候進行處理
1.2 使用場景
1：Building real-time streaming data pipelines that reliably get data between systems or applications.在系統或應用程序之間構建可靠的用於傳輸實時數據的管道，消息隊列功能
2：Building real-time streaming applications that transform or react to the streams of data。構建實時的流數據處理程序來變換或處理數據流，數據處理功能
1.3 詳細介紹
Kafka目前主要作為一個分布式的發布訂閱式的消息系統使用，下面簡單介紹一下kafka的基本機制
1.3.1 消息傳輸流程
Producer即生產者，向Kafka集群發送消息，在發送消息之前，會對消息進行分類，即Topic，上圖展示了兩個producer發送了分類為topic1的消息，另外一個發送了topic2的消息。
- Topic即主題，通過對消息指定主題可以將消息分類，消費者可以只關注自己需要的Topic中的消息
- Consumer即消費者，消費者通過與kafka集群建立長連接的方式，不斷地從集群中拉取消息，然后可以對這些消息進行處理。
從上圖中就可以看出同一個Topic下的消費者和生產者的數量並不是對應的。
1.3.2 kafka服務器消息存儲策略
談到kafka的存儲，就不得不提到分區，即partitions，創建一個topic時，同時可以指定分區數目，分區數越多，其吞吐量也越大，但是需要的資源也越多，同時也會導致更高的不可用性，kafka在接收到生產者發送的消息之后，會根據均衡策略將消息存儲到不同的分區中。
在每個分區中，消息以順序存儲，最晚接收的的消息會最后被消費。
1.3.3 與生產者的交互
生產者在向kafka集群發送消息的時候，可以通過指定分區來發送到指定的分區中
也可以通過指定均衡策略來將消息發送到不同的分區中
如果不指定，就會采用默認的隨機均衡策略，將消息隨機的存儲到不同的分區中
1.3.4 與消費者的交互
在消費者消費消息時，kafka使用offset來記錄當前消費的位置
在kafka的設計中，可以有多個不同的group來同時消費同一個topic下的消息，如圖，我們有兩個不同的group同時消費，他們的的消費的記錄位置offset各不相同，不互相干擾。
對於一個group而言，消費者的數量不應該多余分區的數量，因為在一個group中，每個分區至多只能綁定到一個消費者上，即一個消費者可以消費多個分區，一個分區只能給一個消費者消費
因此，若一個group中的消費者數量大於分區數量的話，多余的消費者將不會收到任何消息。
二、Kafka安裝與使用
2.1 下載
你可以在kafka官網 http://kafka.apache.org/downloads下載到最新的kafka安裝包，選擇下載二進制版本的tgz文件，這里我們選擇的版本是2.12-2.1.0，目前的最新版
2.2 安裝
Kafka是使用scala編寫的運行與jvm虛擬機上的程序，雖然也可以在windows上使用，但是kafka基本上是運行在linux服務器上，因此我們這里也使用linux來開始今天的實戰。
首先確保你的機器上安裝了jdk，kafka需要java運行環境，以前的kafka還需要zookeeper，新版的kafka已經內置了一個zookeeper環境，實驗環境可以直接使用
說是安裝，如果只需要進行最簡單的嘗試的話我們只需要解壓到任意目錄即可，這里我們將kafka壓縮包解壓到 / 目錄
2.3 配置
在kafka解壓目錄下下有一個config的文件夾，里面放置的是我們的配置文件
- consumer.properites 消費者配置，這個配置文件用於配置於2.5節中開啟的消費者，此處我們使用默認的即可
- producer.properties 生產者配置，這個配置文件用於配置於2.5節中開啟的生產者，此處我們使用默認的即可
- server.properties kafka服務器的配置，此配置文件用來配置kafka服務器
目前僅介紹幾個最基礎的配置
- broker.id 申明當前kafka服務器在集群中的唯一ID，需配置為integer,並且集群中的每一個kafka服務器的id都應是唯一的，我們這里采用默認配置即可
- listeners 申明此kafka服務器需要監聽的端口號，如果是在本機上跑虛擬機運行可以不用配置本項，默認會使用localhost的地址，如果是在遠程服務器上運行則必須配置，例如：
listeners=PLAINTEXT://192.168.91.129:9092。並確保服務器的9092端口能夠訪問
- zookeeper.connect 申明kafka所連接的zookeeper的地址 ，需配置為zookeeper的地址，如果使用的是kafka高版本中自帶zookeeper，使用默認配置即可。本文使用的是自己搭建的zookeeper，例如： zookeeper.connect=192.168.91.128:2181
基於docker 安裝
環境說明
| 操作系統 | IP地址 | 角色 | 軟件版本 |
| ubuntu-16.04.5-server-amd64 | 192.168.91.128 | zookeeper | 3.4.13 |
| ubuntu-16.04.5-server-amd64 | 192.168.91.129 | Kafka_server | 2.12-2.1.0 |
| ubuntu-16.04.5-server-amd64 | 192.168.91.131 | Kafka_client | 2.12-2.1.0 |
安裝zookeeper
關於 zookeeper 的安裝，請參考鏈接：
https://www.cnblogs.com/xiao987334176/p/10037490.html
安裝kafka_server
新建空目錄
mkdir /opt/kafka_server_base
dockerfile
內容如下：
FROM ubuntu:16.04 # 修改更新源為阿里雲 ADD sources.list /etc/apt/sources.list ADD kafka_2.12-2.1.0.tgz / # 安裝jdk RUN apt-get update && apt-get install -y openjdk-8-jdk --allow-unauthenticated && apt-get clean all EXPOSE 9092 # 添加啟動腳本 ADD run.sh . RUN chmod 755 run.sh ENTRYPOINT [ "/run.sh"]
run.sh
內容如下：
#!/bin/bash if [ -z $zookeeper ];then echo "zookeeper變量不能為空" exit 2 fi cd /kafka_2.12-2.1.0 # 設置zookeeper連接地址 sed -i "123s/localhost/$zookeeper/" /kafka_2.12-2.1.0/config/server.properties # 設置外部訪問地址,綁定域名。注意:此域名必須被docker解析 echo " advertised.listeners=PLAINTEXT://kafka-1.default.svc.cluster.local:9092 " >> /kafka_2.12-2.1.0/config/server.properties # 獲取docker ip ip=`cat /etc/hosts | tail -1 | awk '{print $1}'` # 增加hosts,方便解析域名 echo "$ip kafka-1.default.svc.cluster.local" >> /etc/hosts # 啟動kafka bin/kafka-server-start.sh config/server.properties
注意：
特別要注意 listeners 參數，不能是0.0.0.0，否則啟動報錯！必須是IP地址或者為空。如果是IP地址，客戶端連接時，必須和它是同樣的ip，否則報錯。默認就是保持為空
listeners=PLAINTEXT://:9092
它會監聽本機的所有IP地址。但是，一旦這樣，即使進入docker容器，使用127.0.0.1:9092也無法正常操作。它必須配合另外一個參數advertised.listeners
advertised.listeners 是用來設置外部訪問地址，推薦使用域名。這里使用域名kafka-1.default.svc.cluster.local，注意：使用域名時，docker必須要求能夠解析，否則客戶端操作時，會報錯。
所以在run.sh中，添加了一條hosts記錄。默認情況下，/etc/hosts最后一行的ip就是docker的ip地址。
sources.list
內容如下：
deb http://mirrors.aliyun.com/ubuntu/ xenial main deb-src http://mirrors.aliyun.com/ubuntu/ xenial main deb http://mirrors.aliyun.com/ubuntu/ xenial-updates main deb-src http://mirrors.aliyun.com/ubuntu/ xenial-updates main deb http://mirrors.aliyun.com/ubuntu/ xenial universe deb-src http://mirrors.aliyun.com/ubuntu/ xenial universe deb http://mirrors.aliyun.com/ubuntu/ xenial-updates universe deb-src http://mirrors.aliyun.com/ubuntu/ xenial-updates universe deb http://mirrors.aliyun.com/ubuntu/ xenial-security main deb-src http://mirrors.aliyun.com/ubuntu/ xenial-security main deb http://mirrors.aliyun.com/ubuntu/ xenial-security universe deb-src http://mirrors.aliyun.com/ubuntu/ xenial-security universe
此時，/opt/kafka_server 目錄結構如下：
./ ├── dockerfile ├── kafka_2.12-2.1.0.tgz ├── run.sh └── sources.list
創建鏡像
docker build -t kafka_server_base /opt/kafka_server_base
啟動kafka
docker run -it -p 9092:9092 -e zookeeper=172.17.0.2 kafka_server_base
注意：172.17.0.2 是zookeeper的容器地址
查看9092端口是否啟動
root@jqb-node129:/opt/kafka_danji# netstat -anpt Active Internet connections (servers and established) Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name tcp 0 0 0.0.0.0:22 0.0.0.0:* LISTEN 1392/sshd tcp6 0 0 :::22 :::* LISTEN 1392/sshd tcp6 0 0 :::9092 :::* LISTEN 10291/java
安裝kafka_client
安裝kafka客戶端就比較簡單了，直接解壓壓縮包，使用里面的shell腳本即可，配置文件不需要修改，使用默認即可！
dockerfile
內容如下：
FROM ubuntu:16.04 # 修改更新源為阿里雲 ADD sources.list /etc/apt/sources.list ADD kafka_2.12-2.1.0.tgz / # 安裝jdk RUN apt-get update && apt-get install -y openjdk-8-jdk --allow-unauthenticated && apt-get clean all # 添加啟動腳本 ADD run.sh . RUN chmod 755 run.sh ENTRYPOINT [ "/run.sh"]
run.sh
內容如下：
#!/bin/bash tail -f /kafka_2.12-2.1.0/NOTICE
sources.list
內容如下：
deb http://mirrors.aliyun.com/ubuntu/ xenial main deb-src http://mirrors.aliyun.com/ubuntu/ xenial main deb http://mirrors.aliyun.com/ubuntu/ xenial-updates main deb-src http://mirrors.aliyun.com/ubuntu/ xenial-updates main deb http://mirrors.aliyun.com/ubuntu/ xenial universe deb-src http://mirrors.aliyun.com/ubuntu/ xenial universe deb http://mirrors.aliyun.com/ubuntu/ xenial-updates universe deb-src http://mirrors.aliyun.com/ubuntu/ xenial-updates universe deb http://mirrors.aliyun.com/ubuntu/ xenial-security main deb-src http://mirrors.aliyun.com/ubuntu/ xenial-security main deb http://mirrors.aliyun.com/ubuntu/ xenial-security universe deb-src http://mirrors.aliyun.com/ubuntu/ xenial-security universe
此時 /opt/kafka_client 目錄結構如下：
./ ├── dockerfile ├── kafka_2.12-2.1.0.tgz ├── run.sh └── sources.list
創建鏡像
docker build -t kafka_client /opt/kafka_client
啟動kafka_client
docker run -d -it kafka_client /bin/bash
查看docker進程
root@jqb-node131:/opt/kafka_client# docker ps CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES e07fd7d20814 kafka_client_test "/run.sh /bin/bash" 4 minutes ago Up 4 minutes epic_bardeen
進入 docker
root@jqb-node131:/opt/kafka_client# docker exec -it e07fd7d20814 /bin/bash root@e07fd7d20814:/#
添加一條hosts記錄
echo "172.17.0.2 kafka-1.default.svc.cluster.local" >> /etc/hosts
2.5 創建第一個消息
2.5.1 創建一個topic
Kafka通過topic對同一類的數據進行管理，同一類的數據使用同一個topic可以在處理數據時更加的便捷
創建一個測試topic,名為test,單分區，副本因子是1
root@e07fd7d20814:/# cd /kafka_2.12-2.1.0/ root@e07fd7d20814:/kafka_2.12-2.1.0# bin/kafka-topics.sh --create --zookeeper 192.168.91.128:2181 --topic test --partitions 1 --replication-factor 1 Created topic "test".
在創建topic后可以通過輸入以下命令，來查看已經創建的topic
root@e07fd7d20814:/kafka_2.12-2.1.0# bin/kafka-topics.sh --list --zookeeper 192.168.91.128:2181 test
2.4.2 創建一個消息生產者
打開一個新的終端，重新進入docker
root@jqb-node131:/opt/kafka_client# docker exec -it e07fd7d20814 /bin/bash root@e07fd7d20814:/#
接下來我們創建第一個消息生產者，輸入hello
root@e07fd7d20814:/kafka_2.12-2.1.0# bin/kafka-console-producer.sh --broker-list 192.168.91.129:9092 --topic test >hello >
2.4.3 創建一個消息消費者
root@e07fd7d20814:/kafka_2.12-2.1.0# bin/kafka-console-consumer.sh --bootstrap-server 192.168.91.129:9092 --topic test --from-beginning
消費者創建完成之后，因為還沒有發送任何數據，因此這里在執行后沒有打印出任何數據
不過別着急，不要關閉這個終端，它會一直hold住
在發送完消息之后，可以回到我們的消息消費者終端中，可以看到，終端中已經打印出了我們剛才發送的消息hello
root@e07fd7d20814:/kafka_2.12-2.1.0# bin/kafka-console-consumer.sh --bootstrap-server 192.168.91.129:9092 --topic test --from-beginning hello
kafka錯誤解決
[2018-12-13 14:29:58,771] ERROR Error when sending message to topic test with key: null, value: 4 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for test-0: 1537 ms has passed since batch creation plus linger time
出現上面的錯誤，表示listeners配置和當前連接不匹配。
就如上面提到的，如果listeners指定了監聽ip，客戶端連接時，也必須是這個ip地址，否則會出現上面的提示。
[2018-12-13 14:35:18,017] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 1 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
出現此錯誤，表示客戶端無法解析advertised.listeners配置的域名，這個時候，添加hosts就可以解決了！
三、使用python操作kafka
使用python操作kafka目前比較常用的庫是kafka-python庫
安裝kafka-python
pip3 install kafka-python
生產者
producer_test.py
from kafka import KafkaProducer producer = KafkaProducer(bootstrap_servers='192.168.0.121:9092') # 連接kafka msg = "Hello World".encode('utf-8') # 發送內容,必須是bytes類型 producer.send('test', msg) # 發送的topic為test producer.close()
執行此程序，它沒有輸出！這個是正常的
消費者
from kafka import KafkaConsumer consumer = KafkaConsumer('test', bootstrap_servers=['192.168.0.121:9092']) for msg in consumer: recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value) print(recv)
執行此程序，此時會hold住，因為它在等待生產者發送消息！
再次執行生產者，此時會輸出：
test:0:9: key=None value=b'Hello World'
本文參考鏈接：
https://www.cnblogs.com/hei12138/p/7805475.html