Kafka 被称为下一代分布式消息系统,由 scala 和 Java 编写,是非营利性组织
ASF(Apache Software Foundation,简称为 ASF)基金会中的一个开源项目,比如
HTTP Server、Hadoop、ActiveMQ、Tomcat 等开源软件都属于 Apache 基金会的开
源软件,类似的消息系统还有 RbbitMQ、ActiveMQ、ZeroMQ。
Kafka®用于构建实时数据管道和流应用程序。 它具有水平可伸缩性,容错性, 快速快速性,可在数千家公司中投入生产。
常用消息队列对比:
kafka 最主要的优势是其具备分布式功能、并可以结合 zookeeper 可以实现动态扩 容,Kafka 是一种高吞吐量的分布式发布订阅消息系统。
kafka 优势:
kafka 通过 O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以 TB 的 消息存储也能够保持长时间的稳定性能。 高吞吐量:即使是非常普通的硬件 Kafka 也可以支持每秒数百万的消息。 支持通过 Kafka 服务器分区消息。 支持 Hadoop 并行数据加载。 O(1)就是最低的时空复杂度了,也就是耗时/耗空间与输入数据大小无关,无论 输入数据增大多少倍,耗时/耗空间都不变,哈希算法就是典型的 O(1)时间复杂 度,无论数据规模多大,都可以在一次计算后找到目标
kafka 角色:
Broker:
Kafka 集群包含一个或多个服务器,这种服务器被称为 broker。
Topic :
每条发布到 Kafka 集群的消息都有一个类别,这个类别被称为 topic,(物 理上不同 topic 的消息分开存储在不同的文件夹,逻辑上一个 topic 的消息虽然保 存于一个或多个 broker 上但用户只需指定消息的 topic 即可生产或消费数据而不 必关心数据存于何处),topic 在逻辑上对 record(记录、日志)进行分组保存,消 费者需要订阅相应的 topic 才能消费 topic 中的消息。
Partition :
是物理上的概念,每个 topic 包含一个或多个 partition,创建 topic 时 可指定 parition 数量,每个 partition 对应于一个文件夹,该文件夹下存储该 partition 的数据和索引文件,为了实现实现数据的高可用,比如将分区 0 的数据 分散到不同的 kafka 节点,每一个分区都有一个 broker 作为 leader 和一个 broker 作为 Follower。
分区的优势(分区因子为 3):
一:实现存储空间的横向扩容,即将多个 kafka 服务器的空间结合利用 二:提升性能,多服务器读写 三:实现高可用,分区 leader 分布在不同的 kafka 服务器,比如分区 0 的 leader 为服务器 A,则服务器 B 和服务器 C 为 A 的 follower,而分区 1 的 leader 为服务 器 B,则服务器 A 和 C 为服务器 B 的 follower,而分区 2 的 leader 为 C,则服务 器 A 和 B 为 C 的 follower。
Producer:负责发布消息到 Kafka broker。
Consumer:消费消息,每个 consumer 属于一个特定的 consuer group(可为每个 consumer 指定 group name,若不指定 group name 则属于默认的 group),使用 consumer high level API 时,同一 topic 的一条消息只能被同一个 consumer group 内的一个 consumer 消费,但多个 consumer group 可同时消费这一消息。
kafka 部署:(基于zookeeper)
## 官方教程 https://kafka.apache.org/quickstart # 下载 https://kafka.apache.org/downloads
部署三台服务器的高可用 kafka 环境。
部署环境:
Server1:172.31.2.41 Server2:172.31.2.42 Server3:172.31.2.43
下载
[root@mq1 src]# wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.8.0/kafka_2.12-2.8.0.tgz
创建目录
[root@mq1 src]# mkdir -p /apps [root@mq1 src]# mkdir -p /data/kafka [root@mq2 src]# mkdir -p /apps [root@mq2 src]# mkdir -p /data/kafka [root@mq3 src]# mkdir -p /apps [root@mq3 src]# mkdir -p /data/kafka
解压
[root@mq1 src]# tar xf kafka_2.12-2.8.0.tgz -C /apps/ [root@mq2 src]# tar xf kafka_2.12-2.8.0.tgz -C /apps/ [root@mq3 src]# tar xf kafka_2.12-2.8.0.tgz -C /apps/
做成软链接
[root@mq1 src]# ln -s /apps/kafka_2.12-2.8.0 /apps/kafka [root@mq2 src]# ln -s /apps/kafka_2.12-2.8.0 /apps/kafka [root@mq3 src]# ln -s /apps/kafka_2.12-2.8.0 /apps/kafka
修改配置
[root@mq1 ~]# vim /apps/kafka/config/server.properties broker.id=41 # 唯一的 listeners=PLAINTEXT://172.31.2.41:9092 # 本机的 # 调优 num.network.threads=8 num.io.threads=12 # 数据目录 log.dirs=/data/kafka # zookeeper 集群 zookeeper.connect=172.31.2.41:2181,172.31.2.42:2181,172.31.2.43:2181
以守护进程启动
[root@mq1 ~]# /apps/kafka/bin/kafka-server-start.sh -daemon /apps/kafka/config/server.properties [root@mq2 ~]# /apps/kafka/bin/kafka-server-start.sh -daemon /apps/kafka/config/server.properties [root@mq3 ~]# /apps/kafka/bin/kafka-server-start.sh -daemon /apps/kafka/config/server.properties
查看端口9092
[root@mq1 ~]# ss -tanl LISTEN 0 50 [::ffff:172.31.2.41]:9092 *:*
验证 zookeeper 中 kafka 元数据:
1、Broker 依赖于 Zookeeper,每个 Broker 的 id 和 Topic、Partition 这些元数据信 息都会写入 Zookeeper 的 ZNode 节点中; 2、Consumer 依赖于 Zookeeper,Consumer 在消费消息时,每消费完一条消息, 会将产生的 offset 保存到 Zookeeper 中,下次消费在当前 offset 往后继续消费; ps:kafka0.9 之前 Consumer 的 offset 存储在 Zookeeper 中,kafka0,9 以后 offset 存储在本地。 3、Partition 依赖于 Zookeeper,Partition 完成 Replication 备份后,选举出一个 Leader, 这个是依托于 Zookeeper 的选举机制实现的;
创建 topic:
创建名为 logstashtest,partitions(分区)为 3,replication(每个分区的副本数/每个
分区的分区因子)为 3 的 topic(主题):
在任意 kafaka 服务器操作:
[root@mq1 kafka]# bin/kafka-topics.sh --create --zookeeper 172.31.2.41:2181,172.31.2.42:2181,172.31.2.43:2181 --partitions 3 --replication-factor 3 --topic m66 Created topic m66.
验证 topic:
状态说明:logstashtest 有三个分区分别为 0、1、2,分区 0 的 leader 是 3(broker.id),
分区 0 有三个副本,并且状态都为 lsr(ln-sync,表示可以参加选举成为 leader)。
[root@mq1 kafka]# bin/kafka-topics.sh --describe --zookeeper 172.31.2.41:2181,172.31.2.42:2181,172.31.2.43:2181 --topic m66 Topic: m66 TopicId: NIgq-gm3SFaxsTpvwCoshA PartitionCount: 3 ReplicationFactor: 3 Configs: Topic: m66 Partition: 0 Leader: 43 Replicas: 43,41,42 Isr: 43,41,42 Topic: m66 Partition: 1 Leader: 41 Replicas: 41,42,43 Isr: 41,42,43 Topic: m66 Partition: 2 Leader: 42 Replicas: 42,43,41 Isr: 42,43,41
获取所有 topic:
[root@mq1 kafka]# bin/kafka-topics.sh --list --zookeeper 172.31.2.41:2181,172.31.2.42:2181,172.31.2.43:2181 m66 m66
测试发送消息:
[root@mq1 kafka]# bin/kafka-console-producer.sh --broker-list 172.31.2.41:9092,172.31.2.42:9092,172.31.2.43:9092 --topic m66 >msg1 >msg2 >mgs3
测试获取消息:
可以到任意一台 kafka 服务器测试消息获取,只要有相应的消息获取客户端即可。
[root@mq2 ~]# /apps/kafka/bin/kafka-console-consumer.sh --topic m66 --bootstrap-server 172.31.2.42:9092 --from-beginning msg1 msg1 msg2 mgs3
删除 topic:(必须有的才能删除,没有就会报错)
[root@mq3 src]# /apps/kafka/bin/kafka-topics.sh --delete --zookeeper 172.31.2.41:2181,172.31.2.42:2181,172.31.2.43:2181 --topic m44 Topic m44 is marked for deletion.
范例:
[root@mq3 src]# /apps/kafka/bin/kafka-topics.sh --delete --zookeeper 172.31.2.41:2181,172.31.2.42:2181,172.31.2.43:2181 --topic m10 Error while executing topic command : Topic 'm10' does not exist as expected [2021-08-17 06:30:06,677] ERROR java.lang.IllegalArgumentException: Topic 'm10' does not exist as expected at kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:542) at kafka.admin.TopicCommand$ZookeeperTopicService.deleteTopic(TopicCommand.scala:500) at kafka.admin.TopicCommand$.main(TopicCommand.scala:71) at kafka.admin.TopicCommand.main(TopicCommand.scala) (kafka.admin.TopicCommand$)