安装Kafka 基础命令
目录
解压
改名
修改配置文件
创建目录用于存放日志
创建修改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里面的所有数据