深入了解如何基于Python读写Kafka
作者:Zl_one 发布时间:2021-02-13 09:33:09
这篇文章主要介绍了深入了解如何基于Python读写Kafka,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
本篇会给出如何使用python来读写kafka, 包含生产者和消费者.
以下使用kafka-python客户端
生产者
爬虫大多时候作为消息的发送端, 在消息发出去后最好能记录消息被发送到了哪个分区, offset是多少, 这些记录在很多情况下可以帮助快速定位问题, 所以需要在send方法后加入callback函数, 包括成功和失败的处理
# -*- coding: utf-8 -*-
'''
callback也是保证分区有序的, 比如2条消息, a先发送, b后发送, 对于同一个分区, 那么会先回调a的callback, 再回调b的callback
'''
import json
from kafka import KafkaProducer
topic = 'demo'
def on_send_success(record_metadata):
print(record_metadata.topic)
print(record_metadata.partition)
print(record_metadata.offset)
def on_send_error(excp):
print('I am an errback: {}'.format(excp))
def main():
producer = KafkaProducer(
bootstrap_servers='localhost:9092'
)
producer.send(topic, value=b'{"test_msg":"hello world"}').add_callback(on_send_success).add_callback(
on_send_error)
# close() 方法会阻塞等待之前所有的发送请求完成后再关闭 KafkaProducer
producer.close()
def main2():
'''
发送json格式消息
:return:
'''
producer = KafkaProducer(
bootstrap_servers='localhost:9092',
value_serializer=lambda m: json.dumps(m).encode('utf-8')
)
producer.send(topic, value={"test_msg": "hello world"}).add_callback(on_send_success).add_callback(
on_send_error)
# close() 方法会阻塞等待之前所有的发送请求完成后再关闭 KafkaProducer
producer.close()
if __name__ == '__main__':
# main()
main2()
消费者
kafka的消费模型比较复杂, 我会分以下几种情况来进行说明
1.不使用消费组(group_id=None)
不使用消费组的情况下可以启动很多个消费者, 不再受限于分区数, 即使消费者数量 > 分区数, 每个消费者也都可以收到消息
# -*- coding: utf-8 -*-
'''
消费者: group_id=None
'''
from kafka import KafkaConsumer
topic = 'demo'
def main():
consumer = KafkaConsumer(
topic,
bootstrap_servers='localhost:9092',
auto_offset_reset='latest',
# auto_offset_reset='earliest',
)
for msg in consumer:
print(msg)
print(msg.value)
consumer.close()
if __name__ == '__main__':
main()
2.指定消费组
以下使用pool方法来拉取消息
pool 每次拉取只能拉取一个分区的消息, 比如有2个分区1个consumer, 那么会拉取2次
pool 是如果有消息马上进行拉取, 如果timeout_ms内没有新消息则返回空dict, 所以可能出现某次拉取了1条消息, 某次拉取了max_records条
# -*- coding: utf-8 -*-
'''
消费者: 指定group_id
'''
from kafka import KafkaConsumer
topic = 'demo'
group_id = 'test_id'
def main():
consumer = KafkaConsumer(
topic,
bootstrap_servers='localhost:9092',
auto_offset_reset='latest',
group_id=group_id,
)
while True:
try:
# return a dict
batch_msgs = consumer.poll(timeout_ms=1000, max_records=2)
if not batch_msgs:
continue
'''
{TopicPartition(topic='demo', partition=0): [ConsumerRecord(topic='demo', partition=0, offset=42, timestamp=1576425111411, timestamp_type=0, key=None, value=b'74', headers=[], checksum=None, serialized_key_size=-1, serialized_value_size=2, serialized_header_size=-1)]}
'''
for tp, msgs in batch_msgs.items():
print('topic: {}, partition: {} receive length: '.format(tp.topic, tp.partition, len(msgs)))
for msg in msgs:
print(msg.value)
except KeyboardInterrupt:
break
consumer.close()
if __name__ == '__main__':
main()
关于消费组
我们根据配置参数分为以下几种情况
group_id=None
auto_offset_reset='latest': 每次启动都会从最新出开始消费, 重启后会丢失重启过程中的数据
auto_offset_reset='latest': 每次从最新的开始消费, 不会管哪些任务还没有消费
指定group_id
auto_offset_reset='latest': 从上次提交offset的地方开始消费
auto_offset_reset='earliest': 从上次提交offset的地方开始消费
auto_offset_reset='latest': 只消费启动后的收到的数据, 重启后会从上次提交offset的地方开始消费
auto_offset_reset='earliest': 从最开始消费全量数据
全新group_id
旧group_id(即kafka集群中还保留着该group_id的提交记录)
性能测试
以下是在本地进行的测试, 如果要在线上使用kakfa, 建议提前进行性能测试
producer
# -*- coding: utf-8 -*-
'''
producer performance
environment:
mac
python3.7
broker 1
partition 2
'''
import json
import time
from kafka import KafkaProducer
topic = 'demo'
nums = 1000000
def main():
producer = KafkaProducer(
bootstrap_servers='localhost:9092',
value_serializer=lambda m: json.dumps(m).encode('utf-8')
)
st = time.time()
cnt = 0
for _ in range(nums):
producer.send(topic, value=_)
cnt += 1
if cnt % 10000 == 0:
print(cnt)
producer.flush()
et = time.time()
cost_time = et - st
print('send nums: {}, cost time: {}, rate: {}/s'.format(nums, cost_time, nums // cost_time))
if __name__ == '__main__':
main()
'''
send nums: 1000000, cost time: 61.89236712455749, rate: 16157.0/s
send nums: 1000000, cost time: 61.29534196853638, rate: 16314.0/s
'''
consumer
# -*- coding: utf-8 -*-
'''
consumer performance
'''
import time
from kafka import KafkaConsumer
topic = 'demo'
group_id = 'test_id'
def main1():
nums = 0
st = time.time()
consumer = KafkaConsumer(
topic,
bootstrap_servers='localhost:9092',
auto_offset_reset='latest',
group_id=group_id
)
for msg in consumer:
nums += 1
if nums >= 500000:
break
consumer.close()
et = time.time()
cost_time = et - st
print('one_by_one: consume nums: {}, cost time: {}, rate: {}/s'.format(nums, cost_time, nums // cost_time))
def main2():
nums = 0
st = time.time()
consumer = KafkaConsumer(
topic,
bootstrap_servers='localhost:9092',
auto_offset_reset='latest',
group_id=group_id
)
running = True
batch_pool_nums = 1
while running:
batch_msgs = consumer.poll(timeout_ms=1000, max_records=batch_pool_nums)
if not batch_msgs:
continue
for tp, msgs in batch_msgs.items():
nums += len(msgs)
if nums >= 500000:
running = False
break
consumer.close()
et = time.time()
cost_time = et - st
print('batch_pool: max_records: {} consume nums: {}, cost time: {}, rate: {}/s'.format(batch_pool_nums, nums,
cost_time,
nums // cost_time))
if __name__ == '__main__':
# main1()
main2()
'''
one_by_one: consume nums: 500000, cost time: 8.018627166748047, rate: 62354.0/s
one_by_one: consume nums: 500000, cost time: 7.698841094970703, rate: 64944.0/s
batch_pool: max_records: 1 consume nums: 500000, cost time: 17.975456953048706, rate: 27815.0/s
batch_pool: max_records: 1 consume nums: 500000, cost time: 16.711708784103394, rate: 29919.0/s
batch_pool: max_records: 500 consume nums: 500369, cost time: 6.654940843582153, rate: 75187.0/s
batch_pool: max_records: 500 consume nums: 500183, cost time: 6.854053258895874, rate: 72976.0/s
batch_pool: max_records: 1000 consume nums: 500485, cost time: 6.504687070846558, rate: 76942.0/s
batch_pool: max_records: 1000 consume nums: 500775, cost time: 7.047331809997559, rate: 71058.0/s
'''
来源:https://www.cnblogs.com/zlone/p/12116817.html
猜你喜欢
- 假如你拥有一个庞大的网站,内容又多,那么来访者往往很难找到自己所需要的东东,这时候你就需要一个站内搜索来帮助来访者更快的找到索要的资料了!现
- 开发的很多场景中都会用到手机号的校验和验证码的校验,具体实现如下<div> <input type="text&
- 在服务器部署时,往往都是在后台运行。当程序发生特定的错误时,我希望能够在日志中查询。因此这里熟悉以下 logging 模块的用法。loggi
- 1.func Copy(dst Writer, src Reader) (written int64, err error)这个函数是从一个
- 用程序来求积分的方法有很多,这篇文章主要是有关牛顿-科特斯公式。学过插值算法的同学最容易想到的就是用插值函数代替被积分函数来求积分,但实际上
- Web_THBC 为表示层也就是页面(.aspx) BLL_THBC 为业务逻辑层 DAL_THBC 为数据库交互层 (向数据库执行SQL语
- 一段时间以来,发现有很多人XHTML都不会用,不光是普通的初学者,有的程序员都不是很清楚该怎么写这个XHTML,我这里呢算是把一些常见的应用
- 现在大部分网站都使用asp+access构建,这样的话通过下载access数据库简单就可以对网站进行破坏! 而很多的网站都不太重
- 前言defer语句被用于预定对一个函数的调用。我们把这类被defer语句调用的函数称为延迟函数。而defer 延迟语句在其他编程语言里好像没
- 看代码,再做解释<?php $array=array('a','b','c','
- 之前对bottle做过不少的介绍,也写过一些文章来说明bottle的缺点,最近发现其实之前有些地方说的不太公平,所以趁此机会也来更正一下。&
- Javascript有没有内存泄露?如果有,如何避免?鉴于最近有好几个人问到我类似的问题,看来大家对这部分内容还没有系统的研究过,因此,打算
- asp之家注:有时候我们需要知道我们链接的远程图片是否正常,是否存在,当不存在时如果我们继续引用,就会在网页上留个大大的X,影响了页面美观。
- 小贤是一条可爱的小狗(Dog), 它的叫声很好听(wow), 每次看到主人的时候就会乖乖叫一声(yelp).从这段描述可以得到以下对象:fu
- 下拉框连动JavaScript代码,市区二级联动多级联动下拉选择框,动态获取下一级琥珀无限级联动菜单-JavaScript版 <htm
- 今天有业务需要制作用户头像的需求,在网上找了个可以裁剪大图制作自己希望大小的图片的方法(基于Struts2)。特此记录一下。不废话,具体的步
- 一. 如何调用def f1(arg1, arg2): print('f1', arg1, arg2)def f2
- 这个涉及到的知识点是django数据库查询问题,我们可以在view.py文件中操作blog_list = models.Blog.objec
- 1.django自定义字段类型,实现非主键字段的自增# -*- encoding: utf-8 -*-from django.db.mode
- 在ASP中,如何获得ADO的连接信息? 具体方法见下列代码:<%Sub Connecti