Skip to content

Kafka基本使用

1.安装前的环境准备

shell
/usr/local/kafka/
  • 修改配置文件:/usr/local/kafka/kafka2.11-2.4/config/server.properties
shell
#broker.id属性在kafka集群中必须要是唯一
broker.id= 0
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.65.60:9092
#kafka的消息存储文件
log.dir=/usr/local/data/kafka-logs
#kafka连接zookeeper的地址
zookeeper.connect= 192.168.65.60:2181

2.启动kafka服务器

进入到bin目录下。使用命令来启动

shell
./kafka-server-start.sh -daemon../config/server.properties

验证是否启动成功:

进入到zk中的节点看id是 0 的broker有没有存在(上线)

shell
ls /brokers/ids/

server.properties核心配置详解:

PropertyDefaultDescription
broker.id0每个broker都可以⽤⼀个唯⼀的⾮负整数id进⾏标识;这个id可以作为broker的“名字”,你可以选择任意你喜欢的数字作为id,只要id是唯⼀的即可。
log.dirs/tmp/kafka-logskafka存放数据的路径。这个路径并不是唯⼀的,可以是多个,路径之间只需要使⽤逗号分隔即可;每当创建新partition时,都会选择在包含最少partitions的路径下进⾏。
listenersPLAINTEXT://192.168.65.60:9092server接受客户端连接的端⼝,ip配置kafka本机ip即可
zookeeper.connectlocalhost:2181zooKeeper连接字符串的格式为:hostname:port,此处hostname和port分别是ZooKeeper集群中某个节点的host和port;zookeeper如果是集群,连接⽅式为hostname1:port1, hostname2:port2,hostname3:port3
log.retention.hours168每个⽇志⽂件删除之前保存的时间。默认数据保存时间对所有topic都⼀样。
num.partitions1创建topic的默认分区数
default.replication.factor1⾃动创建topic的默认副本数量,建议设置为⼤于等于2
min.insync.replicas1当producer设置acks为-1时,min.insync.replicas指定replicas的最⼩数⽬(必须确认每⼀个repica的写数据都是成功的),如果这个数⽬没有达到,producer发送消息会产⽣异常
delete.topic.enablefalse是否允许删除主题

3.创建主题topic

topic是什么概念?topic可以实现消息的分类,不同消费者订阅不同的topic。

输入图片说明

执行以下命令创建名为“test”的topic,这个topic只有一个partition,并且备份因子也设置为1

shell
./kafka-topics.sh --create --zookeeper 192.168.253.131:2181 --replication-factor 1 --partitions 1 --topic test

查看当前kafka内有哪些topic

shell
./kafka-topics.sh --list --zookeeper 192.168.253.131:2181

4.发送消息

kafka自带了一个producer命令客户端,可以从本地文件中读取内容,或者我们也可以以命令行中直接输入内容,并将这些内容以消息的形式发送到kafka集群中。在默认情况下,每一个行会被当做成一个独立的消息。使用kafka的发送消息的客户端,指定发送到的kafka服务器地址和topic

shell
./kafka-console-producer.sh --broker-list 192.168.253.131:9092 --topic test

5.消费消息

对于consumer,kafka同样也携带了一个命令行客户端,会将获取到内容在命令中进行输 出, 默认是消费最新的消息 。使用kafka的消费者消息的客户端,从指定kafka服务器的指定 topic中消费消息

方式一:从最后一条消息的偏移量+1开始消费

shell
./kafka-console-consumer.sh --bootstrap-server 192.168.253.131:9092 --topic test

方式二:从头开始消费

shell
./kafka-console-consumer.sh --bootstrap-server 192.168.253.131:9092 --from-beginning --topic test

几个注意点:

  • 消息会被存储
  • 消息是顺序存储
  • 消息是有偏移量的
  • 消费时可以指明偏移量进行消费