一、Kafka的基础
1、什么是kafka？
Kafka是一个由Apache软件基金会开发的分布式流处理平台。它最初是为LinkedIn设计的，用于处理大规模的实时日志数据流。Kafka的设计目标是将实时流数据高效地发布、订阅和处理。
Kafka是一个基于发布-订阅模式的消息队列系统，可以用于构建实时数据管道和流式处理应用程序。它采用可水平扩展的架构，能够处理大规模的消息流，并保证高吞吐量和低延迟。
Kafka的核心概念包括以下几个部分：
- Topic（主题）：消息发布的类别或名称。
- Producer（生产者）：负责向指定的Topic发布消息。
- Consumer（消费者）：从指定的Topic订阅并消费消息。
- Broker（代理）：Kafka集群中的一台或多台服务器，负责存储和分发消息。
- Partition（分区）：每个Topic可以被划分为多个分区，每个分区在存储层面上对应一个文件夹。
- Offset（偏移量）：用于标识消息在分区中的位置。
Kafka具有高可靠性、可伸缩性和持久性的特点，适用于大规模的数据处理场景，如日志收集、用户活动跟踪、消息系统等。它被广泛应用于各个领域，包括大数据处理、实时数据分析和流式数据处理等。
2、kafka的意义？
将大量数据请求通过Kafka推送到MySQL数据库和直接将数据写入MySQL中，各有优缺点。以下是对两种方式的分析：
-
先推到Kafka，然后再从Kafka消费数据存到MySQL数据库： 优点：
- 异步处理：将数据发送到Kafka之后，可以立即返回响应给客户端，而不需要等待数据写入数据库完成。这样可以提高接口的响应速度和吞吐量。
- 数据解耦：通过使用Kafka作为中间件，可以将数据源和数据消费者解耦。数据生产者只关心将数据发送到Kafka，而数据消费者只关心从Kafka中获取数据进行处理，两者之间没有直接的依赖关系。
- 容错性：Kafka具有高可用性和数据冗余的特性，即使其中一个节点故障，也能保证数据不会丢失。
缺点：
- 增加系统复杂性：引入Kafka作为中间件，会增加系统的复杂性和维护成本。需要管理Kafka集群和监控其状态，以确保数据流畅地传输。
- 需要额外的处理逻辑：在消费Kafka中的数据并存储到MySQL之前，需要编写额外的消费逻辑来处理数据的转换和验证。
-
直接将数据写入MySQL中： 优点：
- 简单直接：直接将数据写入MySQL，省去了推送到Kafka和消费Kafka的额外逻辑和维护成本。
- 实时性：数据可以实时写入MySQL数据库，消费者可以立即从数据库中读取最新的数据。
缺点：
- 同步阻塞：大量的数据写入可能会导致接口阻塞，影响系统的响应速度和吞吐量。客户端需要等待数据写入完成才能收到响应。
- 数据耦合：直接将数据写入MySQL，数据源和数据消费者之间存在紧密的耦合关系。如果有其他数据消费者需要使用这些数据，需要直接访问MySQL数据库。
基于以上考虑，综合来看，如果业务场景对接口响应速度和高吞吐量要求较高，并且希望实现数据解耦和容错性，可以选择先推到Kafka再消费存入MySQL的方式。如果对实时性要求较高，并且不需要处理大量的并发请求，可以直接将数据写入MySQL中。具体选择哪种方式还需要根据具体业务需求、系统架构和性能要求进行评估和权衡。
3、kafka的生命周期？
Kafka是一个高性能、可扩展的分布式流处理平台，用于处理实时数据流。在Kafka中，数据的生产和消费具有以下生命周期：
- 创建Topic：首先需要创建一个Topic，它是数据发布和订阅的基本单元。可以使用Kafka的命令行工具或API来创建Topic。
- 生产者发送数据：生产者将数据发布到指定的Topic中。生产者可以使用Kafka提供的客户端API来发送数据。数据被拆分成多个分区，并按照指定的分区策略写入不同的分区。
- 数据写入分区：一旦生产者发送数据，Kafka会根据指定的分区策略将数据写入相应的分区。分区策略可以是轮询、随机、按键等方式。
- 数据保留：Kafka会根据配置的数据保留策略来决定保留多长时间的数据。可以设置数据保留的时间或者通过设置数据的大小来限制数据保留。
- 消费者订阅数据：消费者可以订阅一个或多个Topic，以接收数据。消费者也可以使用Kafka提供的客户端API来订阅Topic。
- 从分区中读取数据：一旦消费者订阅了Topic，它会从各个分区中拉取数据。消费者可以通过指定偏移量来读取数据，偏移量表示数据在分区中的位置。
- 数据消费确认：消费者成功处理数据后，可以向Kafka提交消费确认。Kafka会记录消费确认的偏移量，并在下次拉取数据时从该偏移量开始读取。
- 数据保留和清理：Kafka会根据配置的数据保留策略来删除过期的数据。已经被消费确认的数据将被删除，以释放存储空间。
这就是Kafka中一条数据的产生与消费的生命周期。通过这个过程，Kafka可以实现高吞吐量、可靠性和可扩展性的数据流处理。
4、kafka的数据写入分区方式 -分区策略
当生产者发送数据到Kafka时，Kafka会根据事先指定的分区策略，将数据写入相应的分区。分区策略决定了数据应该被写入哪个分区。
分区策略可以有不同的方式。例如：
-
轮询策略会依次将数据写入每个分区，保持各个分区数据的平衡；
-
随机策略会随机选择一个分区进行写入；
-
按键策略会根据消息的键值（Key）来确定分区，相同键值的消息将被写入到同一个分区中。
通过选择不同的分区策略，可以灵活地控制数据在Kafka分区之间的分布，以满足特定的需求，比如负载均衡、顺序性等。这样可以更好地利用Kafka的并发处理能力和容错性，提高整个系统的性能和可靠性。
5、kafka的数据保留
数据保留：Kafka会根据配置的数据保留策略来决定保留多长时间的数据。可以设置数据保留的时间或者通过设置数据的大小来限制数据保留。
在Kafka中，可以通过配置数据保留策略来决定保留多长时间的数据。数据保留策略有两种方式可以进行配置：基于时间和基于大小。 基于时间的数据保留策略： 可以通过以下配置参数来设置数据保留的时间： log.retention.ms - 这个参数表示一个日志片段（log segment）最长保留的时间，单位是毫秒。默认值为7天。 log.retention.minutes - 这个参数表示一个日志片段最长保留的时间，单位是分钟。默认值为null。 log.retention.hours - 这个参数表示一个日志片段最长保留的时间，单位是小时。默认值为24小时。 你可以根据需求设置其中的一个参数，或者同时设置多个参数来限制数据保留的时间。 基于大小的数据保留策略： 可以通过以下配置参数来设置数据保留的大小： log.retention.bytes - 这个参数表示一个主题的所有分区在磁盘上占用的最大字节数。默认值为-1，表示没有限制。 log.segment.bytes - 这个参数表示一个日志片段的大小，当到达这个大小时，将会创建一个新的日志片段。默认值为1GB。 你可以根据需求设置其中的一个参数，或者同时设置多个参数来限制数据保留的大小。 例如，如果你想设置一个主题的数据保留时间为3天，可以在配置文件（如server.properties）中添加以下配置： log.retention.hours=72 同样地，如果你想设置一个主题的数据保留大小为10GB，可以在配置文件中添加以下配置： log.retention.bytes=-1 log.segment.bytes=10737418240 配置文件中的参数设置完毕后，重启Kafka服务使其生效。 注意：数据保留策略的配置会影响磁盘空间的使用和数据的可用性，请根据实际需求谨慎配置。
6、kafka中一个topic被消费完后，然后另一个消费者也想消费这个topic，会从哪个分区开始消费呢
在 Kafka 中，每个消费者组都可以独立地消费一个或多个主题的消息。当一个主题被消费完后，如果有新的消费者加入消费者组并且希望消费该主题，它将会从最早的可用分区开始消费。 Kafka的分区是主题的部分副本，每个分区有一个连续的偏移量。当一个主题被完全消费时，消费者组中的所有消费者都会记录每个分区的最后一个消费的偏移量。当新的消费者加入消费者组时，它会查询每个分区的最后一个偏移量，并从最早的可用分区开始消费。 因此，当一个主题被消费完后，新加入的消费者将会从第一个可用分区开始消费，即从已有的最早偏移量开始消费。这样可以确保每个消费者都能按照顺序消费主题的消息，并且可以有效地进行负载均衡。 需要注意的是，如果你想要多个消费者同时消费同一个分区，那么你需要将它们加入到同一个消费者组中。每个消费者组中的消费者将会以协作的方式共同消费分区的消息。
7、kafka的消费机制
Kafka是一种分布式流平台，用于发布和订阅消息流。在Kafka中，消费者可以以消费组的形式进行组织，每个消费组可以有多个消费者实例。消费者通过订阅主题（topic）来接收消息。 Kafka的消费机制基于消费者记录偏移量（offset）来实现。偏移量是一个表示消息在分区中位置的唯一标识符。消费者会周期性地将已消费的消息的偏移量提交给Kafka，在下一次消费时，会从上一次提交的偏移量开始继续消费。 具体来说，Kafka提供了两种方式记录消费偏移量：自动提交和手动提交。 自动提交：消费者可以配置自动提交偏移量的间隔时间。一旦消费者成功消费了一批消息，就会自动提交这批消息最大的偏移量。这种方式简单方便，但可能存在一些消息重复消费或丢失的风险。 手动提交：消费者可以通过编程方式手动提交偏移量。消费者在消费一批消息后，可以显式地调用提交偏移量的API。这种方式可以确保消费者对消息的处理和偏移量的提交是原子性的，从而更加可靠。 在手动提交偏移量时，可以选择同步提交或异步提交。同步提交会阻塞消费者线程，直到收到提交响应；异步提交不会阻塞消费者线程，但需要注意处理提交异常的情况。
总之，Kafka的消费机制通过记录偏移量来实现消息的可靠消费。消费者可以选择自动提交或手动提交偏移量，以满足不同的需求和场景。
try: consumer = Consumer({'group.id': config.topic_gat1400_person_buffer, 'bootstrap.servers': config.bootstrap_servers, 'enable.auto.commit': False, 'auto.offset.reset': 'latest'}) consumer.subscribe([config.topic_gat1400_person_buffer]) producer = KafkaProducer( bootstrap_servers=config.bootstrap_servers, max_request_size=config.max_request_size, key_serializer=str.encode, api_version=(0, 10) ) except Exception as e: log.error("[进程 {}] 初始化异常: {}".format(i, e)) log.info("处理人体记录服务启动 {}".format(os.getpid())) while 1: try: start_time = time.time() for message in consumer.consume(): msg_start_time = time.time() # log.info("[进程 {}] Person处理进度: topic={}, key={}, timestamp={}, offset={}".format( # i, # message.topic, # str(message.key(), encoding="utf-8"), # message.timestamp, # message.offset) # ) params = json.loads(message.value().decode('utf-8')) consumer.commit(asynchronous=False) 解释： 这段代码是Kafka的消费机制基于消费者记录偏移量（offset）来实现。偏移量是一个表示消息在分区中位置的唯一标识符。消费者会周期性地将已消费的消息的偏移量提交给Kafka，在下一次消费时，会从上一次提交的偏移量开始继续消费。 在这段代码中，首先创建了一个消费者对象，并配置了与消费相关的参数，比如消费组ID、启动服务器地址等。然后通过subscribe方法订阅了指定的主题。接着创建了一个生产者对象，用于发送消息。如果出现异常，则会记录错误日志并结束程序。 在while循环中，通过consumer.consume()方法获取到消息，然后解析消息内容并进行相应的处理。处理完成后，使用consumer.commit()方法提交已消费的消息偏移量。这里通过设置asynchronous参数为False，表示同步提交，即会等待提交成功的响应。 这样，每次启动程序时，消费者会从上一次提交的偏移量开始继续消费消息，确保不会漏掉任何消息。Kafka的消费者组参数可以设置为以下两个：
- enable.auto.commit: 这个参数用于控制消费者是否自动提交消费位移。当设置为True时，消费者会定期自动提交消费位移；当设置为False时，消费者需要手动调用commit()方法来提交位移。在一些特殊场景下，我们可能希望精确控制消费位移的提交时机，这时可以将enable.auto.commit设置为False，并手动提交位移。
- auto.offset.reset: 这个参数用于指定当消费者初次订阅一个Topic时，或者消费位移失效（例如因为消费者长时间不活跃）时，如何重置消费位移。它可以设置为以下三个值：
- latest: 默认值，表示重置消费位移为最新的消息。
- earliest: 表示重置消费位移为最早的消息。
- none: 表示如果没有找到消费位移，则抛出异常。
对于给出的参数配置，enable.auto.commit被设置为False，表示消费者不会自动提交消费位移；而auto.offset.reset被设置为latest，表示消费者在初次订阅Topic时，或者消费位移失效时，会重置消费位移为最新的消息。
在消费者组里创建多个消费者
from confluent_kafka import Consumer, KafkaError import threading def consume_messages(consumer_id): consumer = Consumer({ 'bootstrap.servers': 'your_bootstrap_servers', 'group.id': 'your_consumer_group_id', 'enable.auto.commit': False, 'auto.offset.reset': 'latest' }) consumer.subscribe(['your_topic']) while True: message = consumer.poll(1.0) if message is None: continue if message.error(): if message.error().code() == KafkaError._PARTITION_EOF: continue else: print('Error: {}'.format(message.error())) break # 处理消息的逻辑 print('Consumer {}: Received message: {}'.format(consumer_id, message.value().decode('utf-8'))) consumer.commit(message) # 创建多个消费者线程 num_consumers = 3 consumers = [] for i in range(num_consumers): t = threading.Thread(target=consume_messages, args=(i,)) consumers.append(t) t.start() # 等待所有消费者线程结束 for consumer in consumers: consumer.join()注意：在消费者组里的多个消费者消费的数据是不一样的。在Kafka中，每个分区的消息只能被同一个消费者组中的一个消费者处理。当有多个消费者时，Kafka会自动将分区分配给消费者，并确保每个消费者只处理被分配的分区。因此，多个消费者可以并行地处理不同的分区的消息，从而提高处理效率。
6、kafka的运行机制和体系构架
Kafka是一种分布式流数据平台，具有高吞吐量、低延迟、可扩展性和容错性等特点。它的运行机制和体系架构如下：
- 数据模型：Kafka将数据以消息（Message）的形式进行组织和存储。每个消息由一个键（Key）和一个值（Value）组成，同时还包含了时间戳（Timestamp）和其他元数据。
- 主题（Topic）：消息被发布到主题中，主题是消息的分类单位。可以理解为一个逻辑上的消息队列，多个生产者可以将消息发布到同一个主题，多个消费者可以订阅并从主题中消费消息。
- 分区（Partition）：每个主题可以被分为多个分区，分区是Kafka实现高吞吐量和可扩展性的关键。每个分区都是有序的、不可变的消息序列，在分区内保证了消息的顺序。分区还可以水平扩展，使得Kafka可以处理大规模的数据。
- 生产者（Producer）：生产者负责向主题发布消息。它可以选择将消息发送到特定的分区，也可以由Kafka自动为消息选择分区。生产者还可以设置消息的键，这样相同键的消息将被发送到同一个分区，保证这些消息的顺序。
- 消费者（Consumer）：消费者可以订阅一个或多个主题，并从中消费消息。每个消费者都有一个唯一的消费者组（Consumer Group）标识，消费者组内的消费者共同消费主题中的消息。Kafka保证同一个分区内的消息只会被同一个消费者组中的一个消费者消费，保证了消息的负载均衡。
- 消费者偏移量（Consumer Offset）：消费者将记录自己消费的消息位置，即消费者偏移量。Kafka通过保存消费者组的偏移量来实现消费者的故障恢复和负载均衡。
- 日志存储：Kafka使用日志（Log）的形式来存储消息，日志是一个有序的、不断追加的文件。Kafka的高吞吐量得益于顺序写磁盘的性能优势。此外，Kafka还支持数据的保留策略，可以配置数据在磁盘上的保存时间和大小。
- ZooKeeper：作为Kafka的协调服务，ZooKeeper用于管理Kafka集群的元数据和状态信息。Kafka依赖ZooKeeper进行分布式协调和故障恢复。
总之，Kafka通过分布式的方式进行数据的高效传输和持久化存储，适用于大规模的数据流处理场景，如日志收集、实时数据管道、流式处理等。它的体系架构和运行机制的设计使得Kafka具备了高可靠性、高吞吐量和可扩展性等特点。
猜你喜欢
网友评论
- 搜索
- 最新文章
- 热门文章