iis服务器助手广告广告
返回顶部
首页 > 资讯 > 后端开发 > Python >python多线程性能测试之快速mock数据测试分析
  • 716
分享到

python多线程性能测试之快速mock数据测试分析

2023-07-02 08:07:14 716人浏览 泡泡鱼

Python 官方文档:入门教程 => 点击学习

摘要

今天小编给大家分享一下python多线程性能测试之快速mock数据测试分析的相关知识点,内容详细,逻辑清晰,相信大部分人都还太了解这方面的知识,所以分享这篇文章给大家参考一下,希望大家阅读完这篇文章后有所收获,下面我们一起来了解一下吧。安装

今天小编给大家分享一下python多线程性能测试之快速mock数据测试分析的相关知识点,内容详细,逻辑清晰,相信大部分人都还太了解这方面的知识,所以分享这篇文章给大家参考一下,希望大家阅读完这篇文章后有所收获,下面我们一起来了解一下吧。

安装相关类包

  • pip install kafka

  • pip install appmetrics

  • pip install faker

  • pip install pykafka

快速 mock kafka 批量测试数据

# -* coding:utf8 *-from pykafka import KafkaClientimport uuidimport timeimport threadingfrom appmetrics import metricsfrom faker import Fakerimport osfake = Faker("zh-cn")PATH = lambda p: os.path.abspath(    os.path.join(os.path.dirname(__file__), p))meter = metrics.new_meter("meter_test")host_producer = 'host地址'def data_info():    uid = str(uuid.uuid4())    suid = ''.join(uid.split('-'))    return suiddef data_result():    #数据格式可自行定义    data = f"{data_info()},{fake.phone_number()},111111111111,LOL-UZI"    return datadef mock_request():    client_producer = KafkaClient(hosts=host_producer)    topicdocu = client_producer.topics['XXXXXXX-TOPIC']    producer = topicdocu.get_producer(sync=False) # sync=False  关闭同步,使用异步    while True:        data_uni = data_result()        producer.produce(bytes(data_uni, encoding='utf-8'))        meter.notify(1) # 请求一次 记录器打点一次        # i = i - 1    producer.stop()def print_meter():    while True:        print(meter.get())        time.sleep(1)def thread_request(nums):    t1 = []    for i in range(nums):        if i == 0:            #该线程是为了记录每秒打点作用            t = threading.Thread(target=print_meter, name="T" + str(i))        else:            t = threading.Thread(target=mock_request, name="T" + str(i))        t.setDaemon(True)        t1.append(t)    for t in t1:        t.start()    for t in t1:        t.join()##if __name__ == '__main__':    thread_request(101)

appmetrics 使用方法

Meters

Meters,度量一系列事件发生的速率 (rate),例如 TPS。Meters 会统计最近 1 分钟,5 分钟,15 分钟,还有全部时间的速率。

meter = metrics.new_meter(“meter_test”)meter.notify(1)meter.notify(1)meter.notify(3)meter.get()

返回结果:

{'count': 5, 'kind': 'meter', 'five': 0.0066114184713530035, 'mean': 0.27743058841197027, 'f

以上就是“Python多线程性能测试之快速mock数据测试分析”这篇文章的所有内容,感谢各位的阅读!相信大家阅读完这篇文章都有很大的收获,小编每天都会为大家更新不同的知识,如果还想学习更多的知识,请关注编程网Python频道。

--结束END--

本文标题: python多线程性能测试之快速mock数据测试分析

本文链接: https://www.lsjlt.com/news/340553.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑,方便收藏和打印~

下载Word文档
软考高级职称资格查询
编程网,编程工程师的家园,是目前国内优秀的开源技术社区之一,形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容,为IT开发者提供了一个发现、使用、并交流开源技术的平台。
  • 官方手机版

  • 微信公众号

  • 商务合作