一、下载的kafka，部署方式
1、 下载Apache Kafka
2、 安装
2.1解压 tar -zxvf kafka_2.13-3.2.1.tgz
注意解压后的目录需要更改名称 mv kafka-3.2.1-src kafka
2.2 创建数据存放目录
2.2.1先进入kafka 主目录 cd /home/server/kafak
2.2.2 创建存放数据的目录
mkdir -p zk/data
2.2.3 创建上传上传数据时的日志目录
mkdir -p zk/logs
2.2.4 备份默认zookeeper 配置
mv zookeeper.properties zookeeper.properties.bak
2.2.5 新建zookeeper 配置文件
cat > zookeeper.properties << EOF
tickTime=2000
dataDir=/home/server/kafka/zk/data
dataLogDir=/home/server/kafka/zk/logs
clientPort=2181
EOF
2.3.6 配置kafka节点
先创建kafka运行日志存放目录
mkdir -p kafka/logs
cd config
mv server.properties server.properties.bak
[root@localhost config]# cat > server.properties << EOF
> broker.id=1
> 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=/home/server/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
> EOF
说明：上边配置的server.properties的内容基本都是原server.properties的默认配置，到自己安装时主要修改：
broker.id--broker的id；修改为任意自己想要的数值（和zookeeper中的id类似的）
listeners--监听址址；修改为kafka要监听的地址，并且需要在/etc/hosts中配置对应的转发地址，本地搭建的可以直接写localhost，否则在创建topic的时候会报下图错误
log.dirs--日志文件存放目录；修改为自己要存放日志的目录
zookeeper.connect--zookeeper监听地址；修改为自己的zookeeper的监听地址，如果是集群所有地址全写上用逗号（半角）隔开即可,同listeners类似，需要在/etc/hosts中配置项对应的映射
2.3.7 配置kafak环境变量
vim /etc/profile
120 #kafka
121 export KAFKA_HOME=/home/server/kafka
122 export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
2.3.8 配置kafka服务
vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/home/server/kafka'
case $1 in
start)
echo "---------- Kafka 启动 ------------"
${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
echo "---------- Kafka 停止 ------------"
${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
$0 stop
$0 start
;;
status)
echo "---------- Kafka 状态 ------------"
count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
if [ "$count" -eq 0 ];then
echo "kafka is not running"
else
echo "kafka is running"
fi
;;
*)
echo "Usage: $0 {start|stop|restart|status}"
esac
2.3.9 启动
chmod +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start
ps -ef | grep kafka
安装gradle
unzip gradle-7.5.1-bin.zip
mv gradle-7.5.1 gradle
配置环境变量
export GRADLE_HOME=/home/server/gradle
export PATH=$PATH:$GRADLE_HOME/bin
2.4 使用2.2 及更高版本创建topic
./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic heima --partitions 2 --replication-factor 1
说明: --bootstrap-server localhost:9092相当于- -zookeeper localhost:2181
注意：使用2.2 之前版本的kafka时创建topic的方式有所不同 ./kafka-topics.sh --create --zookeeper 192.168.0.103:2181 --replication-factor 2 --partitions 3 --topic heima
老版本说明:
--zookeeper 定义 zookeeper 集群服务器地址，如果有多个 IP 地址使用逗号分割，一般使用一个 IP 即可
--replication-factor 定义分区副本数，1 代表单副本，建议为 2
--partitions 定义分区数
--topic 定义 topic 名称
2.5 使用2.2及以上版本查看
./kafka-topics.sh --describe --bootstrap-server localhost:9092
注意：使用老版本的kafka查询topic时语法稍有不同kafka-topics.sh --describe --zookeeper 192.168.253.44:2181 topic详情
2.6 发送消息
./kafka-console-producer.sh --topic heima --bootstrap-server localhost:9092
注意：老版本发送消息为：./kafka-console-producer.sh --broker-list localhost:9092
2.7 消费消息
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic heima --from-beginning
说明：1、2以及中间的字符都是发送的消息
2.8 修改分区数
./kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic heima --partitions 6
注意：2.2以及其他之前的老版本的修改分区数的方式如下
./kafka-topics.sh --zookeeper localhost:9092 --alter --topic heima --partitions 6
2.9 删除topic
./kafka-topics.sh –delete –bootstrap-server localhost:9092 –topic test
注意：老版本的kafka删除topic为：
./kafka-topics.sh --delete --zookeeper 192.168.253.44:2181,192.168.253.66:2181,192.168.253.77:2181 --topic test
