python多线程性能测试之快速mock数据
目录
- 背景
- 安装相关类包
- 快速 mock kafka 批量测试数据
- appmetrics 使用方法
- Meters
背景
在我们测试工作中,性能测试也是避免不了的,因此在性能测试前期准备工作中,需要 mock 足够批量的数据进行压测。那么怎么能在短时间内快速 mock 出想要的格式数据和足够量的数据进行压测?那么往下看。
安装相关类包
- pip install kafka
- pip install appmetrics
- pip install faker
- pip install pykafka
快速 mock kafka 批量测试数据
# -* coding:utf8 *- from pykafka import KafkaClient import uuid import time import threading from appmetrics import metrics from faker import Faker import os fake = 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 suid def data_result(): #数据格式可自行定义 data = f"{data_info()},{fake.phone_number()},111111111111,LOL-UZI" return data def 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测试多线程mock数据的资料请关注我们其它相关文章!
赞 (0)