消息队列简介
学习 Kafka 不可避免地要认识下消息队列，也就是我们常提到的 MQ(Message Queue)，因为 Kafka 本质上也是一个消息队列。那么消息队列又是什么呢？先来看一个比较官方的回答。
消息队列是一种进程间通信或者同一个进程中不同线程间的通信方式，主要解决异步处理、应用耦合、流量消峰等问题，实现高性能、高可用、可伸缩和最终一致性架构，是大型分布式系统不可缺少的中间件。
再说的直观点，如下图，系统 A 将消息发布到 MQ，然后系统 B 再从 MQ 取消息。
那么，为什么系统 A 不直接发消息给系统 B ，中间还要隔一个 MQ 呢？这就要看下 MQ 的三个主要功能了。
异步处理
消息队列提供了异步处理机制，因为很多时候用户并不需要立即响应来处理消息，那么通过这个机制就可以把所有消息放入 MQ 中。比如，某系统发来的数据中包含很多图片信息，如果对其中的信息都进行保存处理，用户一番操作下来可能会很久。采用异步处理之后，系统会将所有数据存放在 MQ 中，用户不需要立即处理，大大缩短了系统的响应时间。
应用解耦
消息队列可以对系统间的依赖进行解耦，降低依赖系统变更带来的影响。比如，用户在下单后，订单系统A需要通知系统B、系统C等做出响应的处理。传统的做法，如下图所示。
此时的系统A是强依赖系统B和系统C的，一旦系统B出现故障或者需要重新加入高耦合的系统D时，就必须要更改系统A的代码。
如果经常出现这种依赖系统迭代的情况，那么系统A就会很难维护，可以通过消息队列对依赖系统进行解耦（如下图），这样系统A也无需关心其他系统的可用性。
流量削峰
流量削峰还有个形象的名字叫做削峰填谷，其实就是指当数据量激增时，能够有效地隔离上下游业务，将上游突增的流量缓存起来，真正地填到谷中，以平滑的方式传到下游系统，避免了流量的不规则冲击。
比如，有个活动页面平时也就 50qps，某一特殊时刻访问量突然增多，能达到 1000qps，但是当前系统的处理能力最多为 100qps，这个时候可以通过消息队列来进行削峰填谷，如下图所示。
当然，Kafka 除了以上 MQ 这些功能之外，还提供了消息顺序性保障、回溯消息、持久化存储等功能。
MQ 的两种传输模式
消息在 MQ 中有两种传输模型，分别是点对点(point to point)和发布/订阅(publish/subscribe)模型。
点对点模型
如图所示，系统A发送的消息只能被系统B接收，其他的任何系统都不能获取到系统A发送的消息。在日常生活中就像A拨通了B的电话，其他人是不可能接听到的。
发布/订阅模型
与点对点模型的区别在于发布/订阅模型多了一个 topic 的概念，可以存在多个发布者向相同主题发送消息，而订阅者也可以存在多个，接收相同主题的消息。在日常生活中就像不同主题的报纸期刊，同时也有不同群体的读者来订阅。
那么 Kafka 属于哪种呢，事实上 Kafka 可以同时支持这两种传输模型。
Kafka简介
Apache Kafka是一个分布式流处理平台，最初由LinkedIn开发，并在2011年成为Apache项目的一部分。Kafka主要用于构建实时数据管道和流应用程序，能够以高吞吐量和低延迟的方式处理数据流。
以下是Kafka的一些核心概念和功能：
核心概念
- Producer（生产者）：负责将数据发布到Kafka集群的一个或多个主题（Topic）中。
- Consumer（消费者）：从Kafka主题中订阅并消费数据。消费者可以单独消费，也可以作为消费组的一部分来消费数据。
- Broker（代理）：Kafka集群中的每个服务器称为一个Broker。它负责接收来自生产者的数据，并将这些数据存储在磁盘上，然后为消费者提供服务。
- Topic（主题）：数据被分类的逻辑渠道。每个主题可以分为多个分区（Partition），从而实现数据的并行处理和负载均衡。
- Partition（分区）：主题的物理分片，每个分区是一个有序的、不可变的记录序列。分区使得Kafka能够实现扩展性和高吞吐量。
- Offset（偏移量）：每个消息在其分区中都有一个唯一的偏移量，标识该消息在分区中的位置。
- Replication（复制）：Kafka支持数据复制，以确保数据的高可用性和容错能力。每个分区可以有多个副本，副本分布在不同的Broker上。
核心功能
- 高吞吐量：Kafka能够处理大量的数据流，每秒可以处理上百万条消息，适合大规模的消息处理。
- 持久性：Kafka通过将消息写入磁盘来保证数据的持久性。
- 可扩展性：Kafka通过增加更多的Broker和分区，能够水平扩展以处理更多的数据和流量。
- 高可用性：通过复制机制，Kafka确保即使在节点故障的情况下也能继续运行。
- 实时处理：支持实时的数据流处理，适用于需要快速响应的数据应用。
- 灵活的发布/订阅模式：支持多种消费模式，包括点对点和发布/订阅。
- 流处理能力：通过Kafka Streams API和其他集成工具（如Apache Flink和Apache Storm），可以对流数据进行复杂的处理和分析。
常见应用场景
- 日志聚合：集中收集和处理来自不同系统和应用程序的日志数据。
- 流分析：实时处理和分析流数据，以便做出快速决策。
- 数据集成：作为数据总线，将数据从不同的源系统整合到数据湖或数据仓库中。
- 事件驱动架构：支持微服务之间的异步通信。
Kafka因其高性能、可扩展性和可靠性而被广泛应用于各种行业和场景，是现代数据架构中的重要组件。
Kafka的架构
Apache Kafka 的架构设计旨在提供高吞吐量、低延迟的实时数据流处理。
核心组件
- Producer（生产者）：生产者是数据的发布者，负责将数据推送到 Kafka 的主题中。生产者可以选择将消息发送到特定的分区，以实现负载均衡。
- Consumer（消费者）：消费者从 Kafka 主题中读取数据。消费者可以单独消费，也可以组成消费组（Consumer Group），在消费组中，每条消息只会被组内的一个消费者处理。
- Broker（代理）：Kafka 集群由多个 Broker 组成。每个 Broker 是一个 Kafka 服务器，负责存储和管理数据。Broker 之间通过 Zookeeper 进行协调，以实现集群管理和 leader 选举。
- Zookeeper：Zookeeper 用于管理 Kafka 集群的元数据，跟踪 Broker 列表、主题和分区信息，以及执行 leader 选举等任务。
- Topic（主题）：主题是数据的逻辑分类单位。生产者将数据发布到主题，消费者从主题订阅数据。主题可以被细分为多个分区。
- Partition（分区）：每个主题可以分为多个分区，分区是消息的存储单位。分区允许 Kafka 实现并行处理和负载均衡。每个分区在其所属的主题中具有唯一的顺序。
- Replica（副本）：为了确保数据的高可用性和容错性，Kafka 支持分区的副本机制。每个分区可以有多个副本，副本分布在不同的 Broker 上。
- Leader 和 Follower：每个分区有一个 leader 和多个 follower 副本。生产者和消费者只与 leader 交互，follower 从 leader 复制数据，以保持一致性。
主题（Topic）
Topic 是 Kafka 中数据的逻辑分类单元，可以理解成一个队列。Broker 是所有队列部署的机器，Producer 将消息发送到特定的 Topic，而 Consumer 则从特定的 Topic 中消费消息。
分区（Partition）
为了提高并行处理能力和扩展性，Kafka 将一个 Topic 分为多个 Partition。每个 Partition 是一个有序的消息队列，消息在 Partition 内部是有序的，但在不同的 Partition 之间没有顺序保证。Producer 可以并行地将消息发送到不同的 Partition，Consumer 也可以并行地消费不同的 Partition，从而提升整体处理能力。
副本（Replica）
每个 Partition 可以有多个副本（Replica），分布在不同的 Broker 上。Kafka 会为分区的多个副本选举一个作为主副本(Leader)，主副本对外提供读写服务，从副本(Follower)实时同步 Leader 的数据。
Kafka 通过副本机制实现高可用性，当一个 Broker 故障时，可以通过副本保证数据不丢失，并继续提供服务。如下图所示，黄色表示 leader，灰色表示 follower。Topic 分了三个 Patition，副本数是 2。
Consumer 和 ConsumerGroup
Kafka 有消费组的概念，每个消费者只能消费所分配到的分区的消息，每一个分区只能被一个消费组中的一个消费者所消费，所以同一个消费组中消费者的数量如果超过了分区的数量，将会出现有些消费者分配不到消费的分区。消费组与消费者关系如下图所示：
数据存储机制
Kafka 的数据存储机制采用了顺序写入磁盘的方式，通过这种方式来提高写入性能。
每个 Partition 的消息被存储在多个 Segment 文件中，每个 Segment 文件由一组连续的消息组成。Segment 文件通过索引和日志文件进行管理，索引文件记录了每条消息在日志文件中的偏移量。
Kafka 的存储机制具备以下几个特点：
- 顺序写入：Kafka 通过顺序写入来提高写入速度和磁盘利用率。
- Segment 文件：消息被分段存储，便于管理和清理。
- 索引机制：通过索引快速定位消息，提高读取效率。
- 日志清理策略：支持基于时间和大小的日志清理策略，确保存储空间的有效利用。
负载均衡
Kafka 的 Topic 可以分成多个 Partition，每个 Paritition 类似于一个队列，单个 Paritition 可以保证数据有序。Kafka 具有优秀的分区分配算法——StickyAssignor，把生产者的消息发送到不同 Paritition，保证 Paritition 的分配尽量地均衡。这样，整个集群的分区尽量地均衡，各个 Broker 和 Consumer 的处理不至于出现太大的倾斜。同一个 Consumer Group 下的 Consumer 并发消费 Paritition，需要注意的是，如果 Consumer Group 下的 Consumer 个数超过 Partition 数量，那么会出现空闲 Consumer。
顺序保证
每个 Kafka 主题（Topic）可以分为多个分区（Partition）。每个分区都是一个有序的、不可变的消息队列。
- 生产者（Producer）将消息发送到分区时，Kafka 按消息的发送顺序将其追加到分区的末尾。
- 消费者（Consumer）读取分区中的消息时，也是按照消息的存储顺序逐条读取。
因此，在同一个分区内，消息的顺序是严格保证的。这对于某些业务场景来说非常重要，特别是需要保证消息顺序性的应用，例如金融交易、订单处理等。
针对消息有序的业务需求，还分为全局有序和局部有序。
- 全局有序：一个 Topic 下的所有消息都需要按照生产顺序消费。
- 局部有序：一个 Topic 下的消息，只需要满足同一业务字段的要按照生产顺序消费。例如：Topic 消息是订单的流水表，包含订单 orderId，业务要求同一个 orderId 的消息需要按照生产顺序进行消费。
全局有序
Kafka 的一个 Topic 可分为多个 Partition，Producer 发送消息的时候，kafka 会使用负载均衡策略将消息发送到其中一个 Partition，会导致顺序是乱的。要保证全局有序，那么一个 Topic 只能存在一个 Partition。 而且对应的 Consumer 也要使用单线程或者保证消费顺序的线程模型。
局部有序
要满足局部有序，只需要在发消息的时候指定 Partition Key，Kafka 对其进行 Hash 计算，根据计算结果决定放入哪个 Partition。这样 Partition Key 相同的消息会放在同一个 Partition，从而保证有序。此时，Partition 的数量仍然可以设置多个，提升 Topic 的整体吞吐量。
高可用性和容错机制
Kafka 通过以下几种机制来实现高可用性和容错性：
- 副本机制：每个 Partition 有多个副本，主副本（Leader）负责读写操作，其它副本（Follower）定期从 Leader 同步数据。当 Leader 发生故障时，会从 Follower 中选举新的 Leader。
- ACK 机制：Producer 发送消息时，可以通过设置 ACK 来确保消息被成功写入 Leader 和 Follower，从而保证数据不丢失。
- ISR（In-Sync Replica）机制：Kafka 维护一个 ISR 列表，记录当前与 Leader 保持同步的副本。只有在 ISR 列表中的副本才会参与 Leader 选举。
- ZooKeeper 协调：Kafka 使用 ZooKeeper 进行分布式协调，管理元数据和集群状态。ZooKeeper 负责管理 Broker 的注册信息、Topic 和 Partition 的元数据以及 Leader 选举等。
消息传递保证
Kafka 提供了三种消息传递保证：
- At most once：消息最多传递一次，可能丢失。
- At least once：消息至少传递一次，可能重复。
- Exactly once：消息准确传递一次，Kafka 在11.0.0 版本引入了事务机制，支持端到端的精确一次语义。
ZooKeeper 的作用
Kafka 将 Broker、Topic 和 Partition 的元数据信息存储在 Zookeeper 上。通过在 Zookeeper 上建立相应的数据节点，并监听节点的变化，Kafka 使用 Zookeeper 完成以下功能：
- 元数据管理：存储 Kafka 的元数据，包括 Broker 列表、Topic 和 Partition 信息、ISR 列表等。
- 分布式协调：负责 Broker 的注册和发现、Leader 选举、负载均衡等。
- 状态监控：监控 Kafka 集群的运行状态，保证系统的一致性和高可用性。
- Broker 注册：Broker 是分布式部署并且之间相互独立，Zookeeper 用来管理注册到集群的所有 Broker 节点。
- Topic 注册：在 Kafka 中，同一个 Topic 的消息会被分成多个分区并将其分布在多个 Broker 上，这些分区信息及与 Broker 的对应关系也都是由 Zookeeper 在维护
- 生产者负载均衡：由于同一个 Topic 消息会被分区并将其分布在多个 Broker 上，因此，生产者需要将消息合理地发送到这些分布式的 Broker 上。
- 消费者负载均衡：与生产者类似，Kafka 中的消费者同样需要进行负载均衡来实现多个消费者合理地从对应的 Broker 服务器上接收消息，每个消费者分组包含若干消费者，每条消息都只会发送给分组中的一个消费者，不同的消费者分组消费自己特定的 Topic 下面的消息，互不干扰。
Kafka 的扩展性
Kafka 的扩展性主要体现在以下几个方面：
- 水平扩展：通过增加 Broker 节点，可以轻松扩展 Kafka 集群的存储和处理能力。
- Partition 扩展：通过增加 Partition 数量，可以提高 Topic 的并行处理能力。
- 动态配置：Kafka 支持在运行时动态调整部分配置，如 Topic 的分区数量和副本因子等。
数据流动
- 生产者发布消息：生产者将消息发送到指定主题的分区中。可以通过轮询或指定分区键的方式将消息分配到不同的分区。
- Broker 存储消息：Broker 接收到消息后，将其持久化到磁盘，并根据配置进行复制，以确保数据的持久性和高可用性。
- 消费者订阅和消费消息：消费者订阅一个或多个主题，并从主题的分区中读取消息。消费者可以是独立的，也可以是消费组的一部分。
- Zookeeper 协调集群：Zookeeper 负责监控 Broker 的状态、管理主题和分区的元数据，以及进行 leader 选举。
Kafka的使用
Kafka的安装
安装 Apache Kafka 包含几个关键步骤，包括环境准备、下载 Kafka、配置和启动服务。
环境准备
Kafka 需要 Java 运行环境。确保已安装 Java Development Kit (JDK) 8 或更高版本。可以通过以下命令检查 Java 版本：java -version
下载和解压 Kafka
下载 Kafka：访问 Apache Kafka 官方下载页面。选择合适的版本并下载 Kafka 二进制文件。
解压文件：使用以下命令解压下载的 Kafka 文件（假设文件名为 kafka_2.13-3.0.0.tgz）：
tar -xzf kafka_2.13-3.0.0.tgz cd kafka_2.13-3.0.0
配置 Zookeeper 和 Kafka
Kafka 依赖 Zookeeper 进行集群管理，通常 Kafka 包含了 Zookeeper 的必要文件。
- 配置 Zookeeper：在config/zookeeper.properties 文件中，可以配置 Zookeeper 的数据存储目录等设置。默认配置通常适用于本地开发环境。
- 配置 Kafka Broker：在config/server.properties 文件中，配置 Broker ID、监听地址、日志目录等。关键配置项包括：
- id：每个 Broker 的唯一标识符。
- dirs：Kafka 日志文件存储路径。
- connect：Zookeeper 的连接字符串，格式为hostname:port。
启动 Zookeeper 和 Kafka 服务
- 启动 Zookeeper：在 Kafka 的根目录下，使用以下命令启动 Zookeeper 服务：bin/zookeeper-server-start.sh config/zookeeper.properties。确保 Zookeeper 正常启动，没有报错。
- 启动 Kafka Broker：在另一个终端窗口中，使用以下命令启动 Kafka Broker：bin/kafka-server-start.sh config/server.properties。确保 Kafka Broker 正常启动，没有报错。
验证安装
- 创建主题：使用 Kafka 提供的命令行工具创建一个主题，验证 Kafka 是否正常工作：bin/kafka-topics.sh –create –topic test-topic –bootstrap-server localhost:9092 –partitions 1 –replication-factor 1
- 列出主题：列出当前 Kafka 中的所有主题，确保刚才创建的主题存在：bin/kafka-topics.sh –list –bootstrap-server localhost:9092
- 启动生产者：使用命令行工具启动一个生产者，向test-topic 发送消息：bin/kafka-console-producer.sh –topic test-topic –bootstrap-server localhost:9092 输入一些消息，按 Enter 发送。
- 启动消费者：使用命令行工具启动一个消费者，从test-topic 读取消息：bin/kafka-console-consumer.sh –topic test-topic –from-beginning –bootstrap-server localhost:9092 确保消费者能接收到生产者发送的消息。
完成这些步骤后，Kafka 应该已在本地环境中成功安装并运行。根据需要，你可以进一步配置 Kafka 和 Zookeeper，以满足生产环境的要求。
如何向kafka推送消息
这里以Python程序为例，要使用 Python 将消息推送到 Kafka，可以使用 kafka-python 库，这是一个流行的 Kafka 客户端库，支持 Kafka 的生产者和消费者功能。
安装 kafka-python
首先，需要安装 kafka-python 库。可以通过 pip 安装：pip install kafka-python
编写生产者代码
以下是一个简单的 Python 生产者示例，演示如何将消息发送到 Kafka：
from kafka import KafkaProducer
# 创建 Kafka 生产者
producer = KafkaProducer(
bootstrap_servers='localhost:9092', # Kafka broker 地址
key_serializer=lambda k: k.encode('utf-8'), # 可选：键序列化
value_serializer=lambda v: v.encode('utf-8') # 值序列化
)
# 发送消息
topic = 'my-topic'
key = 'key1'
value = 'Hello, Kafka with Python!'
producer.send(topic, key=key, value=value)
producer.flush() # 刷新缓冲区，确保所有消息都已发送
# 关闭生产者
producer.close()
print(f'Message with key {key} sent to topic {topic}')
运行生产者
将上述代码保存为一个 Python 文件（例如 kafka_producer.py），然后在终端中运行该文件：python kafka_producer.py
代码说明
- KafkaProducer: 这是kafka-python 提供的用于发送消息的类。
- bootstrap_servers: 指定 Kafka broker 的地址，通常是hostname:port 格式。如果有多个 broker，可以传递一个列表。
- key_serializer 和 value_serializer: 用于将键和值序列化为字节数组。这里使用简单的 UTF-8 编码。
- send: 用于发送消息到指定的主题，可以指定消息的键和值。
- flush: 确保所有缓冲区中的消息都已发送到 Kafka。
- close: 关闭生产者并释放资源。
注意事项
- 主题存在性：确保要发送消息的主题已经在 Kafka 中创建。可以通过 Kafka 命令行工具创建主题。
- 错误处理：在生产环境中，建议添加错误处理和重试机制，以处理可能的网络问题或 Kafka broker 的故障。
- 性能调优：根据需要调整生产者的配置参数，例如批处理大小、压缩类型等，以优化性能。
如何配置Kafka的消息保留策略
在 Apache Kafka 中，消息不会因为被消费而自动删除。Kafka 的设计初衷是为了实现高吞吐量的消息持久化和发布-订阅功能，因此消息的存储与消费是分离的。这意味着即使消息已经被消费者读取，它们仍然会根据主题的保留策略继续存储在 Kafka 中，直到达到配置的保留时间或存储大小限制。
消息保留的关键点：
- 消费与存储分离：消息被消费者读取后，Kafka 不会立即删除消息。消费者通过偏移量（offset）来跟踪读取的位置，而不是通过删除消息来管理已消费的数据。
- 保留策略：消息的删除仅根据主题配置的保留策略（如ms 和 retention.bytes）进行。这些策略决定了消息在 Kafka 中保留的时间或大小限制。例如，如果 retention.ms 设置为7天，消息会在被写入 Kafka 后保留7天，无论是否被消费。
- 多消费者支持：Kafka 支持多个消费者组读取同一个主题。不同的消费者组拥有独立的偏移量，这意味着同一条消息可以被不同的消费者组多次读取，而不影响消息的存储状态。
- 日志压缩（Log Compaction）：对于某些主题，可以启用日志压缩策略。这种策略不会基于时间或大小删除消息，而是保留每个键的最新版本，删除旧版本。这对于需要保留最新状态的场景非常有用。
实际应用中的考虑：
- 数据持久性：如果需要持久化数据以供将来访问或重新处理，可以根据需要调整保留策略。
- 消费者设计：消费者应该设计为能够处理重复消息，因为消息的存在与否与消费状态无关。
- 存储管理：在设置保留策略时，需考虑存储资源的使用，避免因消息无限制保留导致的磁盘耗尽。
配置 Kafka 的消息保留策略主要涉及对主题的保留时间和保留大小进行设置。以下是如何配置 Kafka 消息保留策略的详细步骤和选项：
配置保留时间（retention.ms）
- 描述：ms参数用于设置消息在 Kafka 中保留的时间，单位是毫秒。超过这个时间的消息将被删除。
- 默认值：168小时（7天）
配置步骤：
- 创建主题时设置： 在创建主题时，可以直接指定保留时间。例如，设置消息保留 24 小时：bin/kafka-topics.sh –create –topic my-topic –bootstrap-server localhost:9092 –partitions 1 –replication-factor 1 –config retention.ms=86400000
- 修改现有主题： 对于已经存在的主题，可以使用以下命令修改保留时间：bin/kafka-topics.sh –alter –topic my-topic –bootstrap-server localhost:9092 –config retention.ms=86400000
配置保留大小（retention.bytes）
- 描述：bytes参数用于设置每个分区的最大存储大小。一旦达到这个大小，最旧的消息会被删除。
- 默认值：-1（表示没有大小限制）
配置步骤：
- 创建主题时设置： 在创建主题时，可以指定保留大小。例如，设置每个分区的最大存储大小为 1 GB：bin/kafka-topics.sh –create –topic my-topic –bootstrap-server localhost:9092 –partitions 1 –replication-factor 1 –config retention.bytes=1073741824
- 修改现有主题： 对于已经存在的主题，可以使用以下命令修改保留大小：bin/kafka-topics.sh –alter –topic my-topic –bootstrap-server localhost:9092 –config retention.bytes=1073741824
其他相关配置
- 日志段大小（segment.bytes）：设置每个日志段的大小，默认是 1 GB。当日志段达到这个大小时，Kafka 会开始写入新的日志段。
- 日志段删除延迟（retention.check.interval.ms）：Kafka 定期检查并删除过期的日志段。这个参数指定检查的时间间隔，默认是 5 分钟。
- 日志压缩（Log Compaction）：如果需要基于键保留最新的消息版本，可以启用日志压缩策略。这与时间和大小无关，而是根据键的最新值来删除旧消息。可以通过设置policy=compact 启用日志压缩。
如何使用Python消费kafka消息
编写消费者代码
以下是一个简单的 Python 消费者示例，演示如何从 Kafka 主题中读取消息：
from kafka import KafkaConsumer
# 创建 Kafka 消费者
consumer = KafkaConsumer(
'my-topic', # 要消费的主题
bootstrap_servers='localhost:9092', # Kafka broker 地址
group_id='my-group', # 消费者组 ID
auto_offset_reset='earliest', # 从起始位置开始消费（'latest' 为从最新消息开始）
key_deserializer=lambda k: k.decode('utf-8') if k else None, # 可选：键反序列化
value_deserializer=lambda v: v.decode('utf-8') if v else None # 值反序列化
)
# 读取消息
for message in consumer:
print(f"Received message: {message.value} with key: {message.key} from partition: {message.partition}")
# 注意：此示例中的循环是无限的。根据需要添加退出条件。
将上述代码保存为一个 Python 文件（例如 kafka_consumer.py），然后在终端中运行该文件：python kafka_consumer.py
代码说明
- KafkaConsumer:kafka-python 提供的用于消费消息的类。
- bootstrap_servers: 指定 Kafka broker 的地址，通常是hostname:port 格式。如果有多个 broker，可以传递一个列表。
- group_id: 消费者组 ID。多个消费者在同一个组内可以共同消费一个主题的不同分区。
- auto_offset_reset: 指定消费者在没有初始偏移量或当前偏移量不可用时该如何处理。earliest表示从最早的消息开始消费，latest 表示从最新的消息开始。
- key_deserializer 和 value_deserializer: 用于将字节数组反序列化为字符串。这里使用简单的 UTF-8 解码。
- for message in consumer: 这是一个迭代器，会不断从 Kafka 中读取新消息。
注意事项
- 消费者组：如果多个消费者属于同一个消费者组，它们会共同消费主题的分区。每个分区只能被一个组内的消费者消费。
- 错误处理：在生产环境中，建议添加错误处理机制，以应对可能的网络问题或 Kafka broker 的故障。
- 消息处理：根据需要添加逻辑来处理从 Kafka 中读取的消息，并根据应用需求决定何时退出消费者循环。
在使用 Python 消费 Kafka 中的数据时，通常不需要手动实现复杂的轮询请求。kafka-python 库中的 KafkaConsumer 类已经为你封装了轮询机制，使得消费过程变得简单和高效。
KafkaConsumer 的工作方式
自动轮询：KafkaConsumer 内部实现了自动轮询机制。当你迭代 KafkaConsumer 对象时，它会自动从 Kafka 中获取新消息。这意味着你只需要编写一个简单的循环来处理每条消息，而不需要显式地管理轮询过程。
示例代码：
from kafka import KafkaConsumer
# 创建 Kafka 消费者
consumer = KafkaConsumer(
'my-topic', # 要消费的主题
bootstrap_servers='localhost:9092',
group_id='my-group',
auto_offset_reset='earliest'
)
# 消费消息
for message in consumer:
print(f"Received message: {message.value} with key: {message.key}")
在这个示例中，for message in consumer 这行代码会不断轮询 Kafka 以获取新消息。
配置和优化
- 自动提交偏移量：
- enable_auto_commit=True是默认设置，这意味着消费者会自动提交偏移量。你可以通过 auto_commit_interval_ms 参数来配置提交频率。
- 如果需要更精细的控制，可以设置enable_auto_commit=False，并手动提交偏移量。
- 轮询超时：可以使用poll(timeout_ms=5000) 方法来控制轮询的超时时间。如果在指定时间内没有新消息，poll 会返回一个空的记录集。
- 批量处理：如果需要批量处理消息，可以使用poll 方法一次获取多条消息。
其他注意事项
- 线程安全：KafkaConsumer并不是线程安全的。如果需要在多线程环境中使用，建议为每个线程创建一个独立的消费者实例。
- 资源管理：在使用完消费者后，调用close() 以释放资源。
- 错误处理：在生产环境中，建议添加错误处理机制以应对网络故障或其他异常情况。
参考链接：
