kafka 安装和操作命令

安装kafka及操作命令

先安装jdk

安装kafka的条件：jdk、zookeeper

在新版kafka内置了zookeeper，先安装jdk即可

# ubuntu中安装
apt-get install openjdk-8-jdk
# 安装完后执行 java -version 能看到版本即安装成功

下载kafka

说明：

我的安装下载目录：/usr/local/src

安装目录: /usr/local/kafka

# 官网下载地址：
https://kafka.apachecn.org/downloads.html # 挑最新版下载

# 下载示例：
cd /usr/local/src
wget https://downloads.apache.org/kafka/3.5.0/kafka_2.12-3.5.0.tgz
tar -zxvf kafka_2.12-3.5.0.tgz
mv kafka_2.12-3.5.0 ../kafka
cd ../kafka/

实时运行：
# 先启动zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
或后台运行：nohup bin/zookeeper-server-start.sh config/zookeeper.properties >/dev/null 2>&1 &

# 再启动kafka
bin/kafka-server-start.sh config/server.properties
或后台运行：nohup bin/kafka-server-start.sh config/server.properties >dev>null 2>&1 &

# 关闭
bin/zookeeper-server-stop.sh
bin/kafka-server-stop.sh

后台运行：
# 启动kafka即可，zookeeper是自动运行的
bin/kafka-server-start.sh -daemon config/server.properties

操作命令

# 启动生产者
bin/kafka-console-producer.sh --topic <topic-name> --bootstrap-server <broker-list>
例子:
bin/kafka-console-producer.sh --topic my-topic --bootstrap-server localhost:9092

# 启动消费者
bin/kafka-console-consumer.sh --bootstrap-server <broker-list> --topic <topic-name> --from-beginning --property print.timestamp=true --from-beginning就从最新的开始消费, --property print.timestamp=true 显示入kafka时间
例子:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --property print.timestamp=true
# 指定消费某个分区或偏移量(从第几条开始消费):--partition 2 --offset 3
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --partition 2 --offset 3
# 指定消费时间
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --property print.timestamp=true  | awk -F 'CreateTime:|\t' '$2 >= 1688107994655 && $2 <= 1688107995686 {print $0}'

# 查看所有消费者
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 查看正在消费的组和
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group 1  #1是上一条命令查到的组


# 创建主题：创建一个名为together的主题，有2个分区，每个分区1个副本
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 2 --replication-factor 1 --config retention.ms=259200000

# 查看主题列表
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

# 查看主题详情：查看together主题的详情
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic together

# 删除主题
kafka-topics.sh --delete --topic my-topic --bootstrap-server localhost:9092

报错

1. window下连接虚拟机中的kafka报错：failed to dial: failed to open connection to ubuntu-xenial:9092: dial tcp: lookup ubuntu-xenial: no such host

解决：在hosts中增加ip指向
虚拟机的ip  虚拟机系统名称
如：192.168.4.3 ubuntu-xenial