目录

解压

改名

修改配置文件

创建目录用于存放日志

创建修改myid

添加环境变量

刷新环境变量

测试

启动zookeeper

启动kafka

关闭kafka

kafka基础命令

查看消息队列

创建消息队列

查看队列详情

查询指定队列消息数量

生产者消费者

生产消息

消费消息

---

解压

将kafka压缩包拷贝到虚拟机的 opt/install里面

 

tar zxf kafka_2.12-2.8.0.tgz -C /opt/soft/

改名

mv kafka_2.12-2.8.0/ kafka212

修改配置文件

vim /opt/soft/kafka212/config/server.properties

21    broker.id=0
36    advertised.listeners=PLAINTEXT://192.168.152.192:9092
60    log.dirs=/opt/soft/kafka212/data 消息存放目录
103  log.retention.hours=1680 消息存放时间小时
123  zookeeper.connect=192.168.61.146:2181 连接zookeeper
137  delete.topic.enable=true 设置可以对topic删除，默认不能删除

创建目录用于存放日志

mkdir /opt/soft/kafka212/data

创建修改myid

echo "0">/opt/soft/kafka212/data/myid

添加环境变量

vim /etc/profile

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

 

刷新环境变量

source /etc/profile

测试

启动zookeeper

zkServer.sh start
查看启动状态
zkServer.sh status

启动kafka

方法一：
kafka-server-start.sh /opt/soft/kafka212/config/server.properties
//该命令会使kafka占用一个虚拟机的线程

方法二：
kafka-server-start.sh -daemon /opt/soft/kafka212/config/server.properties
nohup kafka-server-start.sh /opt/soft/kafka212/config/server.properties &
//后台启动kafka，不会占用线程

jps 查看启动状态

关闭kafka

kafka-server-stop.sh

kafka基础命令

查看消息队列

kafka-topics.sh --zookeeper 192.168.15.192:2181 --list

创建消息队列

kafka-topics.sh --create --zookeeper 192.168.152.192:2181 --topic kb21 --partitions 1 --replication-factor 1 //可修改副本和副本

查看队列详情

kafka-topics.sh --describe --zookeeper 192.168.152.192:2181 --topic bigdata

查询指定队列消息数量

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

生产者消费者

生产消息

kafka-console-producer.sh --topic bigdata --broker-list 192.168.152.192:9092

消费消息

kafka-console-consumer.sh --bootstrap-server 192.168.152.192:9092 --topic bigdata
//开启之后可以同步生产消息的输入

kafka-console-consumer.sh --bootstrap-server 192.168.152.192:9092 --topic bigdata --from-beginning
//可以查看存入kafka里面的所有数据