kafka 入门

Posted 2020-12-02 baizhuang

Apache kafka

kafka 是一个分布式的基于发布订阅的消息队列，主要应用于大数据实时处理领域。

什么是消息系统

消息系统负责将数据从一个应用程序传输到另一个应用程序，分布式消息传递基于可靠消息队列的概念。有两种类型的消息模式可用，一种是点对点，

另一种是 发布-订阅（pub-sub）消息系统，大多数是遵循 pub-sub

点对点系统,消息被保存在队列中，一个或者多个消费者可以消耗队列中的消息，但是特定消息只能由某个消费者消费。一旦消费者读取队列中的消息，它

就从队列中消失。

 

 

发布-订阅消息系统，在发布-订阅系统中，消息被保留在主题中，消费者可以订阅一个或者多个主题并使用主题中的消息。在 pub-sub 系统中，消息生产者成为发布者，消息使用者称为订阅者。

 

 

 什么是 Kafka 

Apache Kafka 是一个分布式发布-订阅消息系统和一个强大的队列，可以处理大量的数据，并使消息从一个端点传递到另一个端点。Kafka 消息保留在磁盘上，并在集群内复制以防止数据丢失。

Kafka 的特带：

• 可靠性 kafka 是分布式的，分区，复制，容错
• 可扩展性 
• 耐用性
• 性能

Kafka 工作流程

订阅具有相同 group id 的主题的消费者被认为是单个组，并且消息在他们之间共享。

pub -sub 工作流程

1. 生产者定期向主题发送消息
2. kafka 代理存储该特定主题配置分区中的所有消息，如果生产者发送2个消息并且有2个分区，kafka 将在第一分区中存储第一个消息，第二个分区中存储第二个消息。
3. 消费者订阅特定主题
4. 一旦消费者订阅主题，kafka 将向消费者提供主题当前偏移，并且还将偏移保存在 zookeeper 中
5. 消费者定期请求 kafka 新消息
6. 一旦 kafk 收到再次系统的新消息，将消息转发给消费者
7. 消费者收到消息并进行处理
8. 一旦消息被处理，消费者将向 kafka 代理发送确认
9. 一旦 kafka 收到确认，它将偏移更改为新值，并在 zookeeper 中更新。（0.9版本之后 不再保存在 zk中）
10. 以上流程重复，直到消费者停止请求
11. 消费者可以随时回退/跳到所需的主题偏移量，并订阅后续的消息

Zookeeper 的作用

Apache Kakfa 的一个关键依赖是 Apache Zookeeper ，它是一个分布式配置和同步服务。Zookeeper 是 Kafka 代理和消费者之间的协调接口，Kafka 服务器通过 zookeeper 集群共享信息。

Kafka 在 zookeeper 中存储基本元数据，例如主题、代理、消费者偏移等信息。

由于关键信息存储在 zookeeper 中，并且它通常此整体上复制此数据，因此 kafka 代理 /Zookeeper 的故障不会影响 kafka 集群的状态。一旦 zookeeper 重启，kafka代理之间的领导者选举

也通过使用Zookeeper 在领导者失败的情况下完成。

Kafka 基本操作

启动 zk

bin/zookeeper-server-start.sh config/zookeeper.properties

启动kafka

bin/kafka-server-start.sh config/server.properties

注：JPS 查询进程， QuorumPeerMail 是 Zookeeper 守护进程

创建主题（replication-factor：主题复制因子，数量=broker 数量）

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 
--partitions 1 --topic topic-name

获取主题列表

bin/kafka-topics.sh --list --zookeeper localhost:2181

启动生产者发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name

注：Config / server.properties文件包含代理端口ID，因为我们知道我们的代理正在侦听端口9092，因此您可以直接指定它。

消费者接受消息

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name 
--from-beginning

删除主题

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name

参考：https://www.w3cschool.cn/apache_kafka/