文章目录
1. 消息队列的应用场景
传统的消息队列的主要应用场景包括:缓存/消峰、解耦和异步通信。
① 缓冲/消峰:有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。
② 解耦:允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。
③ 异步通信:允许用户把一个消息放入队列,但并不立即处理它,然后在需要的时候再去处理它们。
2. 消息队列的两种模式
① 点对对模式:消费者主动拉取数据,消息收到后清除消息
② 发布订阅模式:
-
可以有多个topic主题(浏览、点赞、收藏、评论等);
-
消费者消费数据之后,不删除数据;
-
每个消费者相互独立,都可以消费到数据;
3. Kafka 的基础架构
- Producer:消息生产者,就是向 Kafka broker 发消息的客户端。
- Consumer:消息消费者,向 Kafka broker 取消息的客户端。
- Consumer Group:消费者组,由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费;消费者组之间互不影响。所有的消费者都属于某个消费者组,即消费者组是逻辑上的一个订阅者。
- Broker:一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个broker 可以容纳多个 topic。
- Topic:可以理解为一个队列,生产者和消费者面向的都是一个 topic。
- Partition:为了实现扩展性,一个非常大的 topic 可以分布到多个 broker(即服务器)上,一个 topic 可以分为多个 partition,每个 partition 是一个有序的队列。
- Replica:副本。一个 topic 的每个分区都有若干个副本,一个 Leader 和若干个Follower。
- Leader:每个分区多个副本的“主”,生产者发送数据的对象,以及消费者消费数据的对象都是 Leader。
- Follower:每个分区多个副本中的“从”,实时从 Leader 中同步数据,保持和Leader 数据的同步。Leader 发生故障时,某个 Follower 会成为新的 Leader。
4. Kafka 的安装
① http://kafka.apache.org/downloads下载kafka的kafka_2.12-2.3.0.tar
② 修改 zookeeper.properties 文件:
# the directory where the snapshot is stored.
dataDir=F:/install/kafka/Zookeeper/data
③ 修改server.properties文件:
log.dirs=F:/install/kafka/data
④ 在cmd窗口启动Zookeeper服务器:
F:\install\kafka\kafka_2.12-2.3.0>bin\windows\zookeeper-server-start.bat config\zookeeper.properties
zookeeper-server-start.bat :启动zookeeper服务端
config\zookeeper.properties:利用config目录下的这个配置文件来启动
⑤ 重新开启一个cmd窗口启动Kafka服务端:
F:\install\kafka\kafka_2.12-2.3.0>bin\windows\kafka-server-start.bat config/server.properties
kafka-server-start.bat :启动kafka
config/server.properties:利用config目录下的这个配置文件启动
⑥ 重新开启一个cmd使用kafka:
# 创建主题
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 -partitions 1 --topic test
# 查看对应服务器创建的主题
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat --list --bootstrap-server localhost:9092
test
# 消息生产者发送消息
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-console-producer.bat --broker-list localhost:9092 --topic test
>hello
>hello world!
⑦ 上个cmd窗口已被生产者占用,因此再开一个窗口作为消费者消费消息:
# 消费者消费主题test生产的消息
E:\KafKa\kafka_2.12-2.5.0\bin\windows>kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
hello
hello world!
5. Kafka 命令行操作
1. 主题命令行操作
① 创建主题 topic:
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat
# 设置连接kafka broker主机名称和端口号
--bootstrap-server localhost:9092
# 创建主题
--create
# 处置分区数量为1
--partitions 1
# 设置副本数量为1,副本数量需要小于集群服务器数量
--replication-factor 1
# 设置主题的名称
--topic test2
② 查看当前服务器中的所有 topic:
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat
--bootstrap-server localhost:9092
# 查看所有主题
--list
__consumer_offsets
test
test1
test2
③ 查看test 主题的详情:
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat
--bootstrap-server localhost:9092
# 查看主题详细描述
--describe
--topic test
Topic:test PartitionCount:1 ReplicationFactor:1 Configs:segment.bytes=1073741824
Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
④ 修改分区数,分区数只能增加不能减少:
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat
--bootstrap-server localhost:9092
# 修改主题
--alter
--topic test
# 修改分区数
--partitions 2
⑤ 再次查看 test 主题的详情:
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat
--bootstrap-server localhost:9092
--describe
--topic test
Topic:test PartitionCount:2 ReplicationFactor:1 Configs:segment.bytes=1073741824
Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: test Partition: 1 Leader: 0 Replicas: 0 Isr: 0
⑥ 删除 topic :
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat
--bootstrap-server localhost:9092
# 删除主题
--delete
--topic test2
2. 生产者命令行操作
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-console-producer.bat
--broker-list localhost:9092
--topic test
>hello
>haha
>
3. 消费者命令行操作
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-console-consumer.bat
--bootstrap-server localhost:9092
--topic test
--from-beginning
hello
haha