python中kafka的安装和使用教程

1. 什么是KafkaKafka是一个分布式流处理系统流处理系统使它可以像消息队列一样publish(发布)或者subscribe(订阅)消息分布式提供了容错性并发处理消息的机制。kafka运行在集群上集群包含一个或多个服务器。kafka把消息存在topic中每一条消息包含键值key值value和时间戳timestamp。2. kafka基本概念producer: 消息生产者就是向kafka broker发消息的客户端。consumer: 消息消费者是消息的使用方从Kafka Broker 拉取消息负责消费Kafka服务器上的消息。topic: 主题由用户定义并配置在Kafka服务器用于建立Producer和Consumer之间的订阅关系。生产者发送消息到指定的Topic下消息者从这个Topic下消费消息。你可以把它理解为一个队列topic将消息分类生产者和消费者面向的是同一个 topic。partition消息分区一个topic可以分为多个partitionpartition是相对于topic是在在物理上的概念每个partition是一个有序的队列partition中的每条消息都会被分配一个有序的idoffset。broker一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。consumer-group消费者分组用于归组同类消费者。每个consumer属于一个特定的consumer group多个消费者可以共同消息一个Topic下的消息每个消费者消费其中的部分消息这些消费者就组成了一个分组拥有同一个分组名称通常也被称为消费者集群。消费者组内每个消费者负责消费不同分区的数据提高消费能力。一个分区只能由组内一个消费者消费消费者组之间互不影响。所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者。offset消息在partition中的偏移量。每一条消息在partition都有唯一的偏移量消息者可以指定偏移量来指定要消费的消息。记录消费者消费的位置信息监控数据消费到什么位置当消费者挂掉再重新恢复的时候可以从消费位置继续消费。Zookeeper:Kafka集群能够正常工作需要依赖于zookeeperzookeeper 帮助Kafka 存储和管理集群信息。3.安装kafka这里我的安装环境是macbrew install kafkakafka使用zookeeper管理安装过程会自动安装zookeeper以我这里的配置为说明kafka安装路径/opt/homebrew/Cellar/kafka/3.2.1kafka属性文件/opt/homebrew/etc/kafka/server.propertieszookeeper属性文件:/opt/homebrew/etc/kafka/zookeeper.properties修改kafka启动配置文件vi/opt/homebrew/etc/kafka/server.properties该配置为单机版修改kafka的监听地址和端口为localhost:9092。前台方式启动服务/opt/homebrew/opt/kafka/bin/kafka-server-start /opt/homebrew/etc/kafka/server.properties后台方式服务brew services start zookeeper brew services start kafka查看服务启动没4.可视化工具Offset Explorer使用下载地址:https://www.kafkatool.com/download.html下载对应的版本测试链接每个message默认是以二进制形式展示的, 设置为string5.python操作kafka用Python读写Kafka我们要使用的一个第三方库叫做kafka-python。pip install kafka-pythonkafka-python是一个python的Kafka客户端可以用来向kafka的topic发送消息、消费消息。5.1消息生产者代码如下:第一种写法from kafka import KafkaProducer producerKafkaProducer(bootstrap_servers[127.0.0.1:9092],value_serializerlambda m:json.dumps(m,ensure_asciiFalse).encode(),)for_id inrange(1,5):content{title:生命不息运动不止!,index:_id}futureproducer.send(topictest_topic,valuecontent)resultfuture.get(timeout10)print(kafka生产数据:,result)参数解释topic: topic的名称指定向那个topic发送下消息。如果不手动创建topic的话发送数据的时候会自动创建。bootstrap_servers指定kafka服务器接受的类型是数组如果是多台服务器就在数组里面多写几个服务器地址比如[127.0.0.1:9092, 127.0.0.1:9093, 127.0.0.1:9094]value_serializer:用来指定序列化的方式这里使用json来序列化数据从而实现向Kafka传入一个字典Kafka 会自动把value自动转为json格式。第二种写法importjsonimportloggingfromkafkaimportKafkaProducerfromkafka.errorsimportKafkaError logging.basicConfig(levellogging.INFO)loggerlogging.getLogger(__name__)producerKafkaProducer(bootstrap_servers[127.0.0.1:9092],# 生产环境建议写多个 brokeracksall,# 等所有 ISR 确认最安全retries3,linger_ms10,# 攒批提升吞吐batch_size16*1024,key_serializerlambdak:k.encode(utf-8)ifisinstance(k,str)elsek,value_serializerlambdav:json.dumps(v,ensure_asciiFalse).encode(utf-8),)futures[]try:for_idinrange(1,5):content{title:生命不息运动不止!,index:_id}futureproducer.send(topictest_topic,keystr(_id),# 指定 key 保证同一 id 落到同一分区顺序可控valuecontent,# 已配置 value_serializer直接传 dict)# 用回调方式异步处理结果future.add_callback(lambdameta:logger.info(kafka 消息发送成功: topic%s partition%s offset%s,meta.topic,meta.partition,meta.offset,))future.add_errback(lambdaexc:logger.error(kafka 消息发送失败: %s,exc))futures.append(future)finally:producer.flush(timeout10)# 等所有缓冲消息发出去producer.close(timeout10)# 如果需要同步拿到每条结果可以这样# for f in futures:# try:# meta f.get(timeout10)# print(kafka生产消息:, meta.topic, meta.partition, meta.offset)# except KafkaError as e:# logger.error(发送失败: %s, e)这里也可以在发送消息的直接对value用json.dumps()方法。运行结果如下然后在kafka客户端工具offset explorer2看看效果5.2消息消费者此时开另一个程序运行生产者代码即会有输出显示即监听生产者发布的消息只要生产者新产生了一条消息消费端就会接受到一条消息。代码如下:消费方式1 默认消费所有分区的消息import time from kafka import KafkaConsumer bootstrap_servers[127.0.0.1:9092]consumerKafkaConsumer(group_idtest_consumer_group,bootstrap_serversbootstrap_servers)consumer.subscribe(topics[test_topic])formsg in consumer:time.sleep(2)print(-----------------------------------------)print(接受到消息:,msg)print(ftopic {msg.topic})# topicdefaultis stringprint(fpartition {msg.partition})print(fvalue {msg.value.decode()})# bytes to stringprint(ftimestamp {msg.timestamp})print(time ,time.strftime(%Y-%m-%d %H:%M:%S,time.localtime(msg.timestamp/1000)))print(offset偏移量:,msg.offset)运行结果参数解释topictopic的名称表示当前消费者从那个topic监听并读取消息。group_id指定此消费者实例属于的组名group_id这个参数后面的字符串可以任意填写如果两个程序的Topic与group_id相同那么他们读取的数据不会重复两个程序的Topic相同但是group_id不同那么他们各自消费全部数据互不影响auto_offset_rest这个参数有两个值earliest和latest如果省略这个参数那么默认就是latestauto_offset_reset的作用是在你的 group 第一次运行还没有 offset 的时候给你设定初始的 offset。而一旦你这个group已经有offset了那么auto_offset_reset这个参数就不会再起作用了消费方式2, 指定消费分区import time from kafka import KafkaConsumer from kafka import TopicPartition bootstrap_servers[127.0.0.1:9092]consumerKafkaConsumer(group_idtest_consumer_group,bootstrap_serversbootstrap_servers)consumer.assign([TopicPartition(test_topic,0)])formsg in consumer:time.sleep(2)print(接受到消息:,msg)消费方式3手动commit生产中建议使用这种方式import time from kafka import KafkaConsumer from kafka import TopicPartition bootstrap_servers[127.0.0.1:9092]consumerKafkaConsumer(bootstrap_serversbootstrap_servers,consumer_timeout_ms1000,group_idtest_consumer_group,enable_auto_commitFalse,)consumer.assign([TopicPartition(test_topic,0)])formsg in consumer:time.sleep(2)print(接受到消息:,msg)consumer.commit()6.获取所有的topickafka-python获取所有的topic接口是在KafkaConsumer类中实现的fromkafkaimportKafkaConsumer# 获取topic列表以及topic的分区列表defretrieve_topics():consumerKafkaConsumer(bootstrap_serversservers)print(consumer.topics())# 获取topic的分区列表defretrieve_partitions(topic):consumerKafkaConsumer(bootstrap_serversservers)print(consumer.partitions_for_topic(topic))# 获取Consumer Group对应的分区的当前偏移量defretrieve_partition_offset():consumerKafkaConsumer(bootstrap_serversservers,group_idkafka-group-id)tpTopicPartition(kafka-topic,0)consumer.assign([tp])print(starting offset is ,consumer.position(tp))7. 创建、删除topicfromkafkaimportKafkaAdminClientfromkafka.adminimportNewTopicfromkafka.errorsimportTopicAlreadyExistsError adminKafkaAdminClient(bootstrap_serversservers)# 创建topicdefcreate_topic():try:new_topicNewTopic(create-topic,8,3)admin.create_topics([new_topic])exceptTopicAlreadyExistsErrorase:print(e.message)# 删除topicdefdelete_topic():admin.delete_topics([create-topic])8.获取消费组信息fromkafkaimportKafkaAdminClient adminKafkaAdminClient(bootstrap_serversservers)# 获取消费组信息defget_consumer_group():# 显示所有的消费组print(admin.list_consumer_groups())# 显示消费组的offsetsprint(admin.list_consumer_group_offsets(kafka-group-id))9.获取topic配置信息fromkafkaimportKafkaAdminClientfromkafka.adminimportConfigResource,ConfigResourceType adminKafkaAdminClient(bootstrap_serversservers)# 获取topic的配置信息defget_topic_config():resource_configConfigResource(ConfigResourceType.TOPIC,create-topic)config_entriesadmin.describe_configs([resource_config])print(config_entries.resources)10.offset 和 groupID详解10.1consumer消费者kafka消费者可以从多个broker中读取数据也可以消费多个topic中的数据。因为Kafka的broker是无状态的所以consumer必须使用partition offset来记录消费了多少数据。如果一个consumer指定了一个topic的offset意味着该consumer已经消费了该offset之前的所有数据。consumer可以通过指定offset从topic的指定位置开始消费数据。consumer的offset存储在Zookeeper中。10.2 offset偏移量用来保存消费进度。offset表示在当前topic当前groupID下消费到的位置。offset为earliest并不代表offset1.在不进行过期配置的情况下kafka消息默认7天时间就会过期。过期后其offset也就随之发生变化使得用数字进行配置的消费进度并不准确。earliest自动重置到最早的offset。latest看上去重置到最晚的offset。none如果边更早的offset也没有的话就抛出异常给consumer告诉consumer在整个consumer group中都没有发现有这样的offset。10.3 groupID一个字符串用来指示一组consumer所在的组。相同的groupID表示在一个组里。相同的groupID消费记录offset时记录的是同一个offset。所以此处需要注意1如果多个地方都使用相同的groupid可能造成个别消费者消费不到的情况2如果单个消费者消费能力不足的话可以启动多个相同groupid的consumer消费处理相同的逻辑。但是多线程的时候需要增加每个groupid下的partition分区数量便于每个线程稳定读取固定的partition提高消费能力。参考文献https://blog.csdn.net/weixin_43411585/article/details/124774360https://www.cnblogs.com/ExMan/p/14884681.htmlhttps://www.jianshu.com/p/76d4e1eb4882