幫助中心
這里有最新的使用文檔和教程
Centos7系統安裝部署kafka單節點，Centos安裝kafka,kafka安裝
Kafka是由Apache軟件基金會開發的一個開源流處理平臺，由Scala和Java編寫。Kafka是一種高吞吐量的分布式發布訂閱消息系統，它可以處理消費者在網站中的所有動作流數據。
操作系統：CentOS-7.x
kafka版本：kafka_2.12-3.4.0
1、關閉selinux
sed -i 's/^SELINUX=.*/SELINUX=disabled/' /etc/selinux/config
setenforce 0
2、防火墻設置
CentOS-7.x默認使用的是firewall作為防火墻，這里改為iptables防火墻。
kafka默認使用tcp9092端口號
2.1、關閉firewall：
systemctl stop firewalld.service #停止firewall
systemctl disable firewalld.service #禁止firewall開機啟動
systemctl mask firewalld
systemctl stop firewalld
yum remove firewalld
2.2、安裝iptables防火墻
yum install iptables-services #安裝
vi /etc/sysconfig/iptables #編輯防火墻配置文件
# sample configuration for iptables service
# you can edit this manually or use system-config-firewall
# please do not ask us to add additional ports/services to this default configuration
*filter
:INPUT ACCEPT [0:0]
:FORWARD ACCEPT [0:0]
:OUTPUT ACCEPT [0:0]
-A INPUT -m state --state RELATED,ESTABLISHED -j ACCEPT
-A INPUT -p icmp -j ACCEPT
-A INPUT -i lo -j ACCEPT
-A INPUT -p tcp -m state --state NEW -m tcp --dport 22 -j ACCEPT
-A INPUT -p tcp -m state --state NEW -m tcp --dport 9092 -j ACCEPT
-A INPUT -j REJECT --reject-with icmp-host-prohibited
-A FORWARD -j REJECT --reject-with icmp-host-prohibited
COMMIT
:wq! #保存退出
systemctl restart iptables.service #最后重啟防火墻使配置生效
systemctl enable iptables.service #設置防火墻開機啟動
/usr/libexec/iptables/iptables.init restart #重啟防火墻
3、添加hosts解析
vi /etc/hosts #編輯配置文件
192.168.21.128 Kafka001
:wq! #保存退出
4、安裝jdk（kafka依賴軟件）
下載openjdk：https://adoptium.net/zh-CN/temurin/releases
上傳OpenJDK8U-jdk_x64_linux_hotspot_8u362b09.tar.gz到/usr/local/src目錄下
安裝JDK
#創建jdk安裝路徑
mkdir -p /data/server/java
#解壓
tar -zxvf /usr/local/src/OpenJDK8U-jdk_x64_linux_hotspot_8u362b09.tar.gz -C /data/server/java
#進入安裝目錄
cd /data/server/java/jdk8u362-b09
#查看版本信息
./bin/java -version
openjdk version "1.8.0_362"
OpenJDK Runtime Environment (Temurin)(build 1.8.0_362-b09)
OpenJDK 64-Bit Server VM (Temurin)(build 25.362-b09, mixed mode)
#設置環境變量
vi /etc/profile #添加下面的信息
#set java environment
JAVA_HOME=/data/server/java/jdk8u362-b09
JRE_HOME=$JAVA_HOME/jre
CLASS_PATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
export JAVA_HOME JRE_HOME CLASS_PATH PATH
:wq! #保存退出
#讓修改立即生效
source /etc/profile
#查看版本信息
java -version
[root@ZZKG402-D05-P-PODM-C3-ZTE-R8500G4-14-ITCPV jdk8u362-b09]# java -version
openjdk version "1.8.0_362"
OpenJDK Runtime Environment (Temurin)(build 1.8.0_362-b09)
OpenJDK 64-Bit Server VM (Temurin)(build 25.362-b09, mixed mode)
#其他版本的jdk安裝方法相同
5、安裝kafka
5.1、下載地址
https://kafka.apache.org/downloads
https://downloads.apache.org/kafka/3.4.0/kafka_2.12-3.4.0.tgz
上傳到/usr/local/src目錄
5.1創建kafka安裝目錄
mkdir -p /data/server/kafka
cd /usr/local/src
#解壓到安裝目錄
tar zxvf kafka_2.12-3.4.0.tgz -C /data/server/kafka --strip-components 1
5.2配置環境變量
vi /etc/profile
#set kafka environment
export KAFKA_HOME=/data/server/kafka
export PATH=$PATH:$KAFKA_HOME/bin
:wq! #保存退出
source /etc/profile #使配置立即生效
5.3配置zookeeper（kafka依賴軟件）
#創建zookeeper數據目錄
mkdir -p /data/server/kafka/zookeeper
#創建zookeeper日志目錄
mkdir -p /data/server/kafka/log/zookeeper
##進入kafka配置目錄
cd /data/server/kafka/config
cp zookeeper.properties zookeeper.properties.bak #備份
vi zookeeper.properties #編輯修改
dataDir=/data/server/kafka/zookeeper #zookeeper目錄
dataLogDir= /data/server/kafka/log/zookeeper #zookeeper日志目錄
clientPort=2181
maxClientCnxns=10000
tickTime=2000
initLimit=10
syncLimit=5
#metricsProvider.exportJvmInfo=true
4lw.commands.whitelist=*
#添加zookeeper節點
#服務器名稱與地址：集群信息（服務器編號，服務器地址，LF通信端口，選舉端口）
server.1=192.168.21.128:2888:3888
#server.2=192.168.21.129:2888:3888
#server.3=192.168.21.130:2888:3888
:wq! #保存退出
#啟動zookeeper（zookeeper要在kafka前啟動）
cd /data/server/kafka/bin
./zookeeper-server-start.sh -daemon ../config/zookeeper.properties # -daemon：以后臺方式啟動
#查看進程
ps -ef |grep zookeeper
#測試端口
telnet 192.168.21.128 2181
5.4配置kafka
cp /data/server/kafka/config/server.properties /data/server/kafka/config/server.properties.bak
vi /data/server/kafka/config/server.properties #kafka配置文件
broker.id=1 # 唯一標識在集群中的ID，每個節點的broker.id值唯一，1,2,3
listeners=PLAINTEXT://0.0.0.0:9092 #broker 服務器要監聽的地址及端口，默認監聽端口9092
num.network.threads=3 #處理網絡請求的最大線程數
num.io.threads=8 #處理I/O請求的線程數
advertised.listeners=PLAINTEXT://192.168.21.128:9092 #暴露對外訪問的地址 這個是對外提供的地址 , 當client請求到kafka時, 會分發這個地址
log.dirs=/data/server/kafka/kafka-logs
zookeeper.connect=192.168.21.128:2181
#啟動kafka
cd /data/server/kafka/bin
./kafka-server-start.sh -daemon ../config/server.properties # -daemon：以后臺方式啟動
#創建topics
./kafka-topics.sh --create --topic hostScreen --bootstrap-server 192.168.21.128:9092 --replication-factor 1 --partitions 1
#生產測試
./kafka-console-producer.sh --broker-list 192.168.21.128:9092 --topic hostScreen
#消費測試
./kafka-console-consumer.sh --bootstrap-server 192.168.21.128:9092 --topic hostScreen --from-beginning
#查看主題
./kafka-topics.sh --list --bootstrap-server 192.168.21.128:9092
#查看所有消費者組
./kafka-consumer-groups.sh --bootstrap-server 192.168.21.128:9092 --list
5.5、設置zookeeper和kafka開機啟動
vi /etc/rc.d/rc.local
/bin/sh /data/server/kafka/bin/zookeeper-server-start.sh -daemon /data/server/kafka/config/zookeeper.properties
/bin/sh /data/server/kafka/bin/kafka-server-start.sh -daemon /data/server/kafka/config/server.properties
:wq! #保存退出
#默認/etc/rc.local沒有執行權限，需要手動添加執行權限
chmod +x /etc/rc.d/rc.local
sh /data/server/kafka/bin/zookeeper-server-start.sh -daemon /data/server/kafka/config/zookeeper.properties #啟動zookeeper
sh /data/server/kafka/bin/kafka-server-start.sh -daemon /data/server/kafka/config/server.properties #啟動kafka
#關閉kafka
cd /data/server/kafka/bin
./kafka-server-stop.sh
#關閉zookeeper
cd /data/server/kafka/bin
./zookeeper-server-stop.sh
6、相關概念
Broker：kafka節點，多個broker組成kafka集群。
Topic：即主題，kafka通過Topic對消息進行分類，發布到kafka的消息都需要指定Topic。
Producer：即消息生產者，向Broker發送消息的客戶端。
Consumer：即消息消費者，從Broker消費（接收）消息的客戶端。
ConsumerGroup：即消費者組，消費者隸屬于消費者組，同一個分區的消息可以被多個消費者消費，但是同一個消費者組中只能有一個消費者可以消費。
Partition：即分區，每個Topic下都至少有一個分區，分區內部的消息是有序的。
至此，Linux系統下kafka單節點安裝部署完成。