confluentkafka（confluentkafka python）

本篇文章给大家谈谈confluentkafka，以及confluentkafka python对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

• 1、confluent kafka python怎么实时获取数据
• 2、kafka——消费者原理解析
• 3、Kafka中Topic过多异常分析
• 4、如何在kafka-python和confluent-kafka之间做出选择
• 5、基于Kafka的高性能流处理平台——Confluent

confluent kafka python怎么实时获取数据

使用kafkapython读迅悔取实时数据小例子 使用kafkapython读橘昌毕取实时数据小例子圆芹 from kafka import KafkaConsumer from kafka.client import KafkaClient imp

[img]

kafka——消费者原理解析

kafka采用发布订阅模式：一对多。发布订阅模式又分两种：

Kafka为这两种模型提供了单一的消费者抽象模型： 消费者组 （consumer group）。 消费者用一个消费者组名标记自己。 一缺昌个发布在Topic上消息被分发给此消费者组中的一个消费者。 假如所有的消费者都在一个组中，那么这就变成了队列模型。 假如所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型。 一个消费者组中消费者订阅同一个Topic，每个消费者接受Topic的一部分分区的消息，从而实现对消费者的横向扩展，对消息进行分流。

注意：当单个消费者无法跟上数据生成的速度，就可以增加更多的消费者分担负载，每个消费者只处理部分partition的消息，从而实现单个应用程序的横向伸缩。但是不要让消费者的数量多于partition的数量，此时多余的消费者会空闲。此外，Kafka还允许多个应用程序从同一个Topic读取所有的消息，此时只要保证每个应用程序有自己的消费者组即可。

消费者组的概念就是：当有多个应用程序都需要从Kafka获取消息时，让每个app对应一个消费者组，从而使每个应用程序都能获取一个或多个Topic的全部消息；在每个消费者组中，往消费者组中添加消费者来伸缩读取能力和处理能力，消费者组中的每个消费者只处理每个Topic的一部分的消息，每个消费者对应一个线程。

在同一个群组中，无法让一个线程运行多个消费者，也无法让多线线程安全地共享一个消费者。按照规则，一个消费者使用一个线程，如果要在同一个消费者组中运行多个消费者，需要让每个消费者运行在自己的线程中。最好把消费者的逻辑封装在自己的对象中，然后使用java的ExecutorService启动多个线程，使每个消费者运行在自己的线程上,可参考

一个 consumer group 中有多个 consumer，一个 topic 有多个 partition，所以必然会涉及到 partition 的分配问题，即确定哪个 partition 由哪个 consumer 来消费。

关于如何设置partition值需要考虑的因素

Kafka 有两种分配策略，一个是 RoundRobin，一个是 Range，默认为Range，当消费者组内消费者发生变化时，会触发分区分配策略（方法重新分配）。

以上三种现象会使partition的所有权在消费者之间转移，这样的行为叫作再均衡。

再均衡的优点 ：

再均尺慎衡的缺点 ：

RoundRobin 轮询方式将分区所有作为一个整体进行 Hash 排序，消费者组内分配分区个数最大差别为 1，是按照组来分的，可以解决多个消费者消费数据不均衡的问题。

但是，当消费者组内订阅不同主题时，可能造成消费混乱，如下图所示，Consumer0 订阅主题 A，Consumer1 订阅主题 B。

将 A、B 主题的分区排序后分配给消费者组，TopicB 分区中的数据可能 分配到 Consumer0 中。

Range 方式是按照主题来分的，不会产生轮询方式的消费混乱问题。

但是，如下图所示，Consumer0、Consumer1 同时订阅了主题 A 和伏困扒 B，可能造成消息分配不对等问题，当消费者组内订阅的主题越多，分区分配可能越不均衡。

由于 consumer 在消费过程中可能会出现断电宕机等故障，consumer 恢复后，需要从故障前的位置继续消费，所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。

consumer group +topic + partition 唯一确定一个offest

Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始，

consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中，该 topic 为__consumer_offsets。

你如果特别好奇，实在想看看offset什么的，也可以执行下面操作：

修改配置文件 consumer.properties

再启动一个消费者

当消费者崩溃或者有新的消费者加入，那么就会触发再均衡（rebalance），完成再均衡后，每个消费者可能会分配到新的分区，而不是之前处理那个，为了能够继续之前的工作，消费者需要读取每个partition最后一次提交的偏移量，然后从偏移量指定的地方继续处理。

case1：如果提交的偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。

case2：如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。

自动提交的优点是方便，但是可能会重复处理消息

不足：broker在对提交请求作出回应之前，应用程序会一直阻塞，会限制应用程序的吞吐量。

因此，在消费者关闭之前一般会组合使用commitAsync和commitSync提交偏移量。

ConsumerRebalanceListener需要实现的两个方法

下面的例子演示如何在失去partition的所有权之前通过onPartitionRevoked()方法来提交偏移量。

Consumer有个Rebalance的特性，即重新负载均衡，该特性依赖于一个协调器来实现。每当Consumer Group中有Consumer退出或有新的Consumer加入都会触发Rebalance。

之所以要重新负载均衡，是为了将退出的Consumer所负责处理的数据再重新分配到组内的其他Consumer上进行处理。或当有新加入的Consumer时，将组内其他Consumer的负载压力，重新进均匀分配，而不会说新加入一个Consumer就闲在那。

下面就用几张图简单描述一下，各种情况触发Rebalance时，组内成员是如何与协调器进行交互的。

Tips ：图中的Coordinator是协调器，而generation则类似于乐观锁中的版本号，每当成员入组成功就会更新，也是起到一个并发控制的作用。

参考：

Kafka中Topic过多异常分析

1.1 、kafka 中存在一个__consumer_offsets topic 是专门维护所有topic的偏移量，这个topic下面有很多个分区（胡裤默认情况下__consumer_offsets有50个分区），每个topic下的分区节点维护在zk中

这个topic下面有50个分区

每个分区的leader不同，并不是只有一个leader维护这个升绝topic，每个partion都有各自的leader

topic过多裤笑简，导致分区过多，kafka中主要是会受分区数量的影响；

每个Partition都有一个ISR（ISR全称是“In-Sync Replicas”，也就是保持同步的副本，他的含义就是，跟Leader始终保持同步的Follower有哪些。），这个ISR里一定会有Leader自己，因为Leader肯定数据是最新的，然后就是那些跟Leader保持同步的Follower，也会在ISR里。

确切的数字自然依赖于诸如可容忍的不可用窗口时间、Zookeeper延时、broker存储类型等因素。根据经验法则我们评估单台broker能够支撑的分区数可达4000个，而单集群能够支撑200000个分区。当然后者主要受限于集群对controller崩溃这种不常见情形的容忍度，另外其他影响分区数的因素也要考虑进来。

如下链接是说明kafka与分区数量的关系影响

参考链接：

翻译链接：

如何在kafka-python和confluent-kafka之间做出选择

kafka-python：蛮荒的西部薯余

kafka-python是最受欢迎的Kafka Python客户端。我们过衡或去使用时从未出现过任何问题，在我的《敏捷数据科学2.0》一书中我也用过它。然而在最近这个项目中，它却出现了一个严重的问题。我们发现，当以文档化的方式使用KafkaConsumer、Consumer迭代式地从消息队列中获取消息时，最终到达主题topic的由Consumer携带的消息通常会丢失。我们通过控制台Consumer的分析验证了这一点。

需要更详细说明的是，kafka-python和KafkaConsumer是与一个由SSL保护的Kafka服务（如Aiven Kafka）一同使用的，如下面这样：

kafka_consumer = KafkaConsumer(

topic,

enable_auto_commit=True,

group_id=group_id,

bootstrap_servers=config.kafka.host,

api_version=(0, 10),

security_protocol='SSL',

ssl_check_hostname=True,

ssl_cafile=config.kafka.ca_pem,

ssl_certfile=config.kafka.service_cert,

ssl_keyfile=config.kafka.service_key

)

for message in kafka_consumer:

application_message = json.loads(message.value.decode())

...

当以这样的推荐方式使用时，KafkaConsumer会丢失消息。但有一个变通方案，就是保留所有消息。这个方案是Kafka服务提供商Aiven support提供给我们的。它看起来像这样：

while True:

raw_messages = consumer.poll(timeout_ms=1000, max_records=5000)

for topic_partition, messages in raw_messages.items():

application_message = json.loads(message.value.decode())

...

虽然这个变通方案可能有用，但README中的方法会丢弃消息使我对其失去兴趣。所以我找到了一个替代方案。

confluent-kafka：企业支持

发现coufluent-kafka Python模块时，我感到无比惊喜。它既能做librdkafka的外封装，又非常小巧。librdkafka是一个用C语言写的kafka库，它是Go和.NET的基础。更重要的是，它由Confluent公司支持。我爱开源，但是当“由非正式社区拥有或支持”这种方式效果不行的时候，或许该考虑给替代方案印上公章、即该由某个公司拥有或支持了。不过，我们并未购买商业支持。我们知道有人会维护这个库的软件质量，而且可以选择买或不买商业支持，这一点真是太棒了。

用confluent-kafka替换咐手伍kafka-python非常简单。confluent-kafka使用poll方法，它类似于上面提到的访问kafka-python的变通方案。

kafka_consumer = Consumer(

{

"api.version.request": True,

"enable.auto.commit": True,

"group.id": group_id,

"bootstrap.servers": config.kafka.host,

"security.protocol": "ssl",

"ssl.ca.location": config.kafka.ca_pem,

"ssl.certificate.location": config.kafka.service_cert,

"ssl.key.location": config.kafka.service_key,

"default.topic.config": {"auto.offset.reset": "smallest"}

}

)

consumer.subscribe([topic])

# Now loop on the consumer to read messages

running = True

while running:

message = kafka_consumer.poll()

application_message = json.load(message.value.decode())

kafka_consumer.close()

现在我们能收到所有消息了。我并不是说kafka-python工具不好，我相信社区会对它的问题做出反应并解决。但从现在开始，我会一直坚持使用confluent-kafka。

开源治理

开源是强大的，但是涉及到复杂的“大数据”和NoSQL工具时，通常需要有一家大公司在背后推动工具的开发。这样你就知道，如果那个公司可以使用工具，那么该工具应该拥有很好的基本功能。它的出现可能是非正式的，就像某公司发布类似FOSS的项目一样，但也可能是正式的，就像某公司为工具提供商业支持一样。当然，从另一个角度来看，如果一家与开源社区作对的公司负责开发某个工具，你便失去了控制权。你的意见可能无关紧要，除非你是付费客户。

理想情况是采取开源治理，就像Apache基金会一样，还有就是增加可用的商业支持选项。这对互联网上大部分的免费软件来说根本不可能。限制自己只使用那些公司盖章批准后的工具将非常限制你的自由。这对于一些商店可能是正确选择，但对于我们不是。我喜欢工具测试，如果工具很小，而且只专心做一件事，我就会使用它。

信任开源

对于更大型的工具，以上决策评估过程更为复杂。通常，我会看一下提交问题和贡献者的数量，以及最后一次commit的日期。我可能会问朋友某个工具的情况，有时也会在推特上问。当你进行嗅探检查后从Github选择了一个项目，即说明你信任社区可以产出好的工具。对于大多数工具来说，这是没问题的。

但信任社区可能存在问题。对于某个特定的工具，可能并没有充分的理由让你信任社区可以产出好的软件。社区在目标、经验和开源项目的投入时间方面各不相同。选择工具时保持审慎态度十分重要，不要让理想蒙蔽了判断。

基于Kafka的高性能流处理平台——Confluent

软件准备

Confluent平台是一个可靠的，高性能的流处理平台，你可以通过这个平台组织和管理各式各样的数据源中的数据。

confluent start 会启动 confluent 全部组件，如果羡禅想要单独启动，比如单独启动 schema-registry，可以执行以下命令：

具体的单独启动各组件的命令，进入 bin 目录下，一看就能明白，不再赘述。

说明：

confluent 中内嵌了 Kafka 和 Zookeeper，你也可以通过指定不同的 zookeeper 在其他的 kafka 集群中创建 topic 或执行其他操作。

以上命令是内嵌的一个kafka-producer脚本，生成随机的用户信息，可以通过 quickstart=[CLICKSTREAM_CODES, CLICKSTREAM, CLICKSTREAM_USERS, ORDERS, RATINGS, USERS, USERS_, PAGEVIEWS] 来生成不同的数据，这个搏慎脚本会运行很长时间(官网只说了很长时间兄银尘，到底多长，没说)，除非你手动停止

在另一个窗口中，进入KSQL命令行(上一个窗口继续发数据不要停)

把生产过来的数据创建为user表：

设置消费偏移量为 "earliest"：

查询：

关于confluentkafka和confluentkafka python的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。