前置

将压缩包拷贝到虚拟机中

解压

[root@cp145 install]# tar -zxf ./kafka_2.12-2.8.0.tgz -C ../soft/

改名

[root@cp145 soft]# mv kafka_2.12-2.8.0/ kafka212

更改server.properties

[root@cp145 config]# vim ./server.properties 

当前虚拟机地址

消息存放的目录（没有就新建）


日志保存时间（小时）

zookeeper的地址

增加一行，不然保护机制不让删除
可以对topic删除

delete.topic.enable=true

退出后将0输出到消息目录下的myid

[root@cp145 config]# echo "0" > /opt/soft/kafka212/data/myid

添加全局变量

#KAFKA_HOME
export KAFKA_HOME=/opt/soft/kafka212
export PATH=$PATH:$KAFKA_HOME/bin

启动

先启动zookeeper

zkServer.sh start

非静默启动。会占用一个黑窗口

[root@cp145 data]# kafka-server-start.sh /opt/soft/kafka212/config/server.properties 

再启动一个窗口，用jps查看

静默启动 加一个-daemon

[root@cp145 data]# kafka-server-start.sh -daemon /opt/soft/kafka212/config/server.properties

另一种静默启动

[root@cp145 data]# nohup kafka-server-start.sh /opt/soft/kafka212/config/server.properties &

操作命令

创建消息队列

[root@cp145 data]# kafka-topics.sh --create --zookeeper 192.168.153.145:2181 --topic cp --partitions 1 --replication-factor 1
//  --topic cp 起名cp partitions 1 分区数为1 replication-factor 1 副本数为1（备份一次）

查看消息队列

[root@cp145 data]# kafka-topics.sh --zookeeper 192.168.153.145:2181 --list

创建副本（备份）单机版只能备份一个（同一台机器备份多个，没有意义）

–replication-factor 1（有几台机器就写几）

[root@cp145 data]# kafka-topics.sh --create --zookeeper 192.168.153.145:2181 --topic bigdata --partitions 3 --replication-factor 1

查看队列详情

[root@cp145 data]# kafka-topics.sh --describe --zookeeper 192.168.153.145:2181 --topic cp

再查看3个分区的队列

[root@cp145 data]# kafka-topics.sh --describe --zookeeper 192.168.153.145:2181 --topic bigdata

生产消费消息

生产消息

[root@cp145 data]# kafka-console-producer.sh --topic bigdata --broker-list 192.168.153.145:9092

消费消息

[root@cp145 data]# kafka-console-consumer.sh --bootstrap-server 192.168.153.145:9092 --topic bigdata

最后加上–from-beginning，就是从头开始取消息

生产消息输入啥，消费消息就显示啥

查看队列中的消息数量

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.153.145:9092 --topic bigdata

关闭

先关kafka

直接kill -9

再关zookeeper

zkServer.sh stop