本文为博主原创,未经允许不得转载:
1. 安装JDK
由于Kafka是用Scala语言开发的,运行在JVM上,因此在安装Kafka之前需要先安装JDK。
yum install java‐1.8.0‐openjdk* ‐y
2. 安装zookeeper
kafka依赖zookeeper,所以需要先安装zookeeper
2.1 进入zookeeper 官网:https://zookeeper.apache.org/releases.html
选择并下载指定的zookeeper 版本安装包:
2.2 下载并解压
wget https://downloads.apache.org/zookeeper/zookeeper-3.4.11/apache-zookeeper-3.4.11.tar.gz tar ‐zxvf zookeeper‐3.4.11.tar.gz
2.3 配置zookeeper 配置文件,并启动 zookeeper
cp conf/zoo_sample.cfg conf/zoo.cfg bin/zkServer.sh start # 启动zookeeper bin/zkCli.sh ls / #查看zk的根目录相关节点
3. 安装kafka
3.1 下载解压 kafka
打开kafka 官网下载页面 : http://kafka.apache.org/downloads.html , 选择对应的scala 语言版本与kafka版本进行下载:
wget https://archive.apache.org/dist/kafka/2.4.0/kafka_2.12-2.4.0.tgz # 下载 tar ‐xzf kafka_2.12-2.4.0.tgz # 解压
3.2 启动 kafka ,并进入 zookeeper 客户端查看kafka 节点目录
启动脚本语法: kafka-server-start.sh [daemon] server.properties
server.properties 的配置路径是一个强制的参数,daemon表示以后台进程运行,否则ssh客户端退出后,就会停止服务。
bin/kafka-server-start.sh config/server.properties # 直接启动kafka bin/kafka-server-start.sh -daemon config/server.properties # 以后台运行的方式启动kafka bin/zkCli.sh 进入zookeeper目录通过zookeeper客户端查看下zookeeper的目录树 ls / # 查看zk的根目录kafka相关节点 ls /brokers/ids # 查看kafka节点 bin/kafka-server-stop.sh config/server.properties # 停止 kafka
server.properties核心配置详解:
Property |
Default |
Description |
broker.id |
0 |
每个broker都可以用一个唯一的非负整数id进行标识;这个id可以作为broker的“名字”,你可以选择任意你喜欢的数字作为id,只要id是唯一的即可。 |
log.dirs |
/tmp/kafka-logs |
kafka存放数据的路径。这个路径并不是唯一的,可以是多个,路径之间只需要使用逗号分隔即可;每当创建新partition时,都会选择在包含最少partitions的路径下进行。 |
listeners |
PLAINTEXT://192.168.65.60:9092 |
server接受客户端连接的端口,ip配置kafka本机ip即可 |
zookeeper.connect |
localhost:2181 |
zooKeeper连接字符串的格式为:hostname:port,此处hostname和port分别是ZooKeeper集群中某个节点的host和port;zookeeper如果是集群,连接方式为 hostname1:port1, hostname2:port2, hostname3:port3 |
log.retention.hours |
168 |
每个日志文件删除之前保存的时间。默认数据保存时间对所有topic都一样。 |
num.partitions |
1 |
创建topic的默认分区数 |
default.replication.factor |
1 |
自动创建topic的默认副本数量,建议设置为大于等于2 |
min.insync.replicas |
1 |
当producer设置acks为-1时,min.insync.replicas指定replicas的最小数目(必须确认每一个repica的写数据都是成功的),如果这个数目没有达到,producer发送消息会产生异常 |
delete.topic.enable |
false |
是否允许删除主题 |
4. 创建主题
4.1 创建一个名字为“test”的Topic,这个topic只有一个partition,并且备份因子也设置为1:
注意: 当命令执行不通过时,将其中的localhost 更换成服务器的ip,或 127.0.0.1 尝试。 这个ip 是有
#kafka部署的机器ip和提供服务的端口号 决定的: 默认是 127.0.0.1 ,如果指定ip 则以下命令都使用ip访问,
如果没有配置,使用默认配置时,则使用 127.0.0.1 进行访问
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
4.2 查看kafka中目前存在的topic
bin/kafka-topics.sh --list --zookeeper localhost:2181
( 除了我们通过手工的方式创建Topic,当producer发布一个消息到某个指定的Topic,这个Topic如果不存在,就自动创建。)
4.3 删除主题
bin/kafka‐topics.sh ‐‐delete ‐‐topic test ‐‐zookeeper localhost:2181
5. 发送消息
kafka自带了一个producer命令客户端,可以从本地文件中读取内容,或者我们也可以以命令行中直接输入内容,并将这些内容
以消息的形式发送到kafka集群中。在默认情况下,每一个行会被当做成一个独立的消息。首先我们要运行发布消息的脚本,
然后在命令中输入要发送的消息的内容:
bin/kafka‐console‐producer.sh ‐‐broker‐list localhost:9092 ‐‐topic test # 或 localhost 执行时通常需要在服务器的host 文件中配置对应的host才能解析,不然解析不了 bin/kafka-console-producer.sh --broker-list 112.125.26.68:9092 --topic test > produce send a msg
6. 消费消息
对于consumer,kafka同样也携带了一个命令行客户端,会将获取到内容在命令中进行输出,默认是消费最新的消息:
bin/kafka-console-consumer.sh --bootstrap-server 112.125.26.68:9092 --consumer-property group.id=testgroup --topic test
如果想要消费之前的消息可以通过--from-beginning参数指定,如下命令:
bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic test
kafka 使用以上命令消费消息使用有的版本会报: zookeeper is not a recognized option 异常。可以使用以下命令进行消费:
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic wordsendertest --from-beginning
查看组名
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 -list
查看消费者的消费偏移量
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group testgroup
标签: kafka