Kafka的基本操作

1 kafka的topic的操作

​ topic是kafka非常重要的核心概念，是用来存储各种类型的数据的，所以最基本的就需要学会如何在kafka中创建、修改、删除的topic，以及如何向topic生产消费数据。

​ 关于topic的操作脚本：kafka-topics.sh

1.1 创建topic

[root@qf01 kafka-2.4.1]# kafka-topics.sh --create \
--topic hadoop \	## 指定要创建的topic的名称
--zookeeper qf01:2181,qf02:2181,qf03:2181/kafka \ ##指定kafka关联的zk地址
--partitions 3 \		##指定该topic的分区个数
--replication-factor 3	##指定副本因子

注意：指定副本因子的时候，不能大于broker实例个数，否则报错

• zookeeper目录变化

• kafka数据目录的变化

1.2 查看topic的列表

[root@qf01 kafka-2.4.1]# kafka-topics.sh --list --zookeeper qf01:2181,qf02:2181,qf03:2181/kafka
hadoop

1.3 查看每一个topic的信息

[root@qf01 kafka-2.4.1]# kafka-topics.sh --describe --topic hadoop --zookeeper qf01:2181,qf02:2181,qf03:2181/kafka
Topic: hadoop   PartitionCount: 3       ReplicationFactor: 3    Configs:
        Topic: hadoop   Partition: 0    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
        Topic: hadoop   Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
        Topic: hadoop   Partition: 2    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2


Partition:	当前topic对应的分区编号
Replicas :  副本因子，当前kafka对应的partition所在的broker实例的broker.id的列表
Leader	 :  该partition的所有副本中的leader领导者，处理所有kafka该partition读写请求
ISR		 :  该partition的存活的副本对应的broker实例的broker.id的列表

1.4 修改一个topic

[root@qf01 data]# kafka-topics.sh --alter --topic hadoop --partitions 4 --zookeeper qf01:2181,qf02:2181,qf03:2181/kafka

WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

但是注意：partition个数，只能增加，不能减少：

1.5 删除一个topic

[root@qf01 data]# kafka-topics.sh --delete --topic hadoop --zookeeper qf01:2181,qf02:2181,qf03:2181/kafka

Topic hadoop is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

1.6 列举所有topic

[root@qf01 data]# kafka-topics.sh --list --zookeeper qf01:2181,qf02:2181,qf03:2181/kafka

1.7 生产数据

[root@qf01 data]# kafka-console-producer.sh \
--topic hadoop \
--broker-list qf01:9092,qf02:9092,qf03:9092

1.8 消费数据

[root@qf01 ~]# kafka-console-consumer.sh --topic hadoop --bootstrap-server qf01:9092,qf02:9092,qf03:9092

如果在生产者生产的时候没有启动消费者是没数据的，原因在于消费者后于生产者启动，在消费者启动之前生产者消费的数据变不能直接获取。

如果想要获取消费者启动之前生产者生产的数据，可以添加一个参数–from-beginning。

• 如下所示：

[root@qf01 ~]# kafka-console-consumer.sh --topic hadoop --bootstrap-server qf01:9092,qf02:9092,qf03:9092 --from-beginning

333
111
222
444

2 Kafka的数据消费的总结

​ kafka消费者在消费数据的时候，都是分组别的。不同组的消费不受影响，相同组内的消费，需要注意，如果partition有3个，消费者有3个，那么便是每一个消费者消费其中一个partition对应的数据；如果有2个消费者，此时一个消费者消费其中一个partition数据，另一个消费者消费2个partition的数据。如果有超过3个的消费者，同一时间只能最多有3个消费者能消费得到数据。

kafka-console-consumer.sh --topic spark \
 --bootstrap-server qf01:9092,qf02:9092,qf03:9092 \
 --group 1000phone \		---》消费者对应的消费者组
 --offset earliest \		--->从什么位置(消息的偏移量)开始消费
 --partition 2		---->消费哪一个分区中的数据	

offset：是kafka的topic中的partition中的每一条消息的标识，如何区分该条消息在kafka对应的partition的位置，就是用该偏移量。offset的数据类型是Long，8个字节长度。offset在分区内是有序的，分区间是不一定有序。如果想要kafka中的数据全局有序，就只能让partition个数为1。

​ 在组内，kafka的topic的partition个数，代表了kafka的topic的并行度，同一时间最多可以有多个线程来消费topic的数据，所以如果要想提高kafka的topic的消费能力，应该增大partition的个数。