东湖之滨 发表于 2024-8-10 22:17:43

kafka服务介绍

Apache Kafka 是一个开源的分布式事件流平台,主要用于及时数据传输和流处理。它最初由 LinkedIn 开辟,并在 2011 年成为 Apache 基金会的顶级项目。Kafka 计划的目的是处理大规模的数据流,同时提供高吞吐量、低延迟和高容错性
Kafka 的工作原理可以从几个关键方面来明白:


[*]
[*]消息的生产
   

[*]Producer:生产者是发送消息的客户端,向 Kafka 发送消息。生产者将消息发布到特定的主题(Topic)中,主题内部可以有多个分区(Partition)。
[*]Partitioning:消息会被分配到差异的分区。分区是主题的逻辑分片,有助于进步并发处理能力。消息的分配通常基于某种计谋,好比消息键(key)大概轮询。

[*]
[*]消息的存储
   

[*]Broker:Kafka 的服务器组件,负责存储和管理消息。每个 Kafka 实例都是一个 Broker。多个 Brokers 构成 Kafka 集群。
[*]Log Segments:在每个分区中,消息被追加到日记(log)中。日记是一个有序的、不可变的消息序列。为了管理日记文件的大小,Kafka 会定期将日记分段成多个文件。

[*]
[*]消息的消费
   

[*]Consumer:消费者从 Kafka 中读取消息。消费者订阅一个或多个主题,然后拉取(fetch)数据。
[*]Consumer Group:消费者可以构成一个消费组(Consumer Group)。同一消费组中的消费者会分担读取同一个主题的差异分区的任务,从而实现负载均衡。每个消息只会被消费组中的一个消费者读取。

[*]
[*]消息的存储与复制
   

[*]Replication:为了进步数据的可靠性,Kafka 利用副本(replica)。每个分区有一个主副本(leader)和若干个从副本(follower)。所有的读写请求都由主副本处理,从副本则从主副本同步数据。
[*]Leader-Follower:在分区中,领导者负责所有的读写请求,从副本负责从领导者同步数据。领导者失败时,从副本会推举新的领导者,确保高可用性。

[*]
[*]和谐与管理
   

[*]Zookeeper:早期 Kafka 利用 Zookeeper 来管理集群的元数据和和谐集群中的 Broker 和分区状态。Zookeeper 负责领导者的推举、配置管理和状态监控。
[*]Kafka 2.8.0 及以后:Kafka 开始渐渐减少对 Zookeeper 的依赖,尝试用 Kafka 自身的协议来管理集群的元数据,这种模式称为 KRaft 模式(Kafka Raft Metadata Mode)。

数据流示例


[*]生产:生产者将消息发送到一个主题,主题有多个分区。每条消息附带一个时间戳。
[*]存储:消息被追加到分区的日记中。日记分为多个段,Kafka 按次序存储消息。
[*]消费:消费者从主题的分区中读取消息。消费者可以跟踪已处理的消息位置,以实现断点续传。
[*]复制:数据在主副本和从副本之间同步,包管数据在 Broker 失败时不会丢失。
通过这些机制,Kafka 能够实现高吞吐量、低延迟和高可靠性的消息转达和数据流处理。
安装利用

Kafka 依赖于 Java 运行环境,因此首先需要安装 Java 11 或更高版本
apt install -y openjdk-11-jdk
root@huhy:~# java --version
openjdk 11.0.23 2024-04-16
OpenJDK Runtime Environment (build 11.0.23+9-post-Ubuntu-1ubuntu1)
OpenJDK 64-Bit Server VM (build 11.0.23+9-post-Ubuntu-1ubuntu1, mixed mode, sharing)
官网下载;https://downloads.apache.org/kafka/3.7.1/kafka_2.13-3.7.1.tgz
tar -xf kafka_2.13-3.7.1.tgz
cd kafka_2.13-3.7.1/
Kafka 的配置文件位于 config 目次中。主要的配置文件包括:


[*] server.properties:Kafka 的服务器配置文件
[*] zookeeper.properties:Zookeeper 的配置文件(如果利用 Zookeeper)

[*] broker.id:描述:Kafka Broker 的唯一标识符。每个 Broker 必须有一个唯一的 broker.id。
默认值:无
示例:broker.id=0
[*] listeners:描述:Kafka Broker 监听的地点和端口。指定了 Kafka 接收客户端请求的地点。
默认值:PLAINTEXT://:9092
示例:listeners=PLAINTEXT://localhost:9092
[*] advertised.listeners:描述:Kafka 向客户端公开的地点。客户端会通过此地点与 Broker 进行通讯。
默认值:无
示例:advertised.listeners=PLAINTEXT://your-hostname:9092
[*] log.dirs:
描述:Kafka 存储日记文件的目次。可以设置多个目次,Kafka 会将数据分散存储。
默认值:/tmp/kafka-logs
示例:log.dirs=/var/lib/kafka/logs
[*] log.retention.hours:
描述:日记文件的生存时间,单位是小时。超过这个时间的数据会被删除。
默认值:168(7 天)
示例:log.retention.hours=168
[*] log.segment.bytes:
描述:每个日记段的大小,单位是字节。日记文件会被分段存储。
默认值:1073741824(1 GB)
示例:log.segment.bytes=536870912(512 MB)
[*] num.partitions:
描述:默认创建的主题的分区数目。
默认值:1
示例:num.partitions=3
[*] replication.factor:
描述:主题的副本因子,表示每个分区有多少副本。进步副本因子可以增长数据的可靠性。
默认值:无(主题创建时指定)
示例:replication.factor=2
[*] message.max.bytes:
描述:Kafka 答应的最大消息大小,单位是字节。
默认值:1000012(1 MB)
示例:message.max.bytes=2097152(2 MB)
[*] log.retention.bytes:
描述:每个分区的日记文件最大生存大小,超过这个大小的日记会被删除。
默认值:-1(不限制)
示例:log.retention.bytes=1073741824(1 GB)
[*] log.cleaner.enable:
描述:启用日记清算器,用于压缩日记中的重复数据。
默认值:false
示例:log.cleaner.enable=true
[*] security.inter.broker.protocol:
描述:Kafka Broker 之间的通讯协议,支持 PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。
默认值:PLAINTEXT
示例:security.inter.broker.protocol=SSL
[*] ssl.keystore.location:
描述:SSL 密钥库的位置,用于 SSL/TLS 加密。
默认值:无
示例:ssl.keystore.location=/path/to/keystore.jks
[*] zookeeper.connect:
描述:Zookeeper 的连接字符串,包括 Zookeeper 的主机名和端标语。
默认值:localhost:2181
示例:zookeeper.connect=localhost:2181
[*] zookeeper.connection.timeout.ms:
描述:Zookeeper 连接超时设置,单位是毫秒。
默认值:6000
示例:zookeeper.connection.timeout.ms=10000
[*] auto.create.topics.enable:
描述:是否自动创建主题。如果设置为 true,当客户端向不存在的主题发送消息时,Kafka 会自动创建该主题。
默认值:true
示例:auto.create.topics.enable=false
[*] delete.topic.enable:
描述:是否答应删除主题。如果设置为 true,可以通过 Kafka 提供的脚本删除主题。
默认值:false
示例:delete.topic.enable=true

通常环境下,默认配置就可以开始利用。如果需要自界说配置,可以编辑这些文件
启动 Zookeeper;Kafka 需要 Zookeeper 来管理集群的元数据。Kafka 附带了一个简单的 Zookeeper 实例,开启后另起一个终端
bin/zookeeper-server-start.sh config/zookeeper.properties
启动 Kafka Broker;在另一个终端中,启动 Kafka Broker
bin/kafka-server-start.sh config/server.properties
另起一个终端3;Kafka 利用主题来组织消息。可以利用 Kafka 提供的脚本创建主题。例如,创建一个名为 test-topic 的主题:
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
Created topic test-topic. 生产消息:可以利用 Kafka 提供的生产者工具向主题中发送消息。打开一个终端并运行,然后在终端4中输入消息并按回车键发送消息
bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092
>huhy> 消费消息;在另一个终端3中,可以运行消费者工具来读取消息,只有最开始两个终端是不能终端,后两个有交互界面可以直接用
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server localhost:9092
huhy
获取信息如下
https://i-blog.csdnimg.cn/direct/7d0340d210134e84b00a5cbf408a1935.png
管理 Kafka

检察主题:
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsetstest-topic 检察主题详情:
bin/kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092
Topic: test-topic       TopicId: rXHPQIqJRkO5lQDOsco3NQ PartitionCount: 1       ReplicationFactor: 1    Configs:      Topic: test-topic       Partition: 0    Leader: 0       Replicas: 0   Isr: 0 删除主题;
bin/kafka-topics.sh --delete --topic test-topic --bootstrap-server localhost:9092
验证检察
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-topics.sh --delete --topic test-topic --bootstrap-server localhost:9092
root@huhy:~/kafka_2.13-3.7.1# bin/kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsets 停止 Kafka Broker
bin/kafka-server-stop.sh
停止 Zookeeper:
bin/zookeeper-server-stop.sh

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
页: [1]
查看完整版本: kafka服务介绍