python多線程性能測試之快速mock數(shù)據(jù)

更新時(shí)間：2022年06月13日 12:05:09 作者：baoqiang

這篇文章主要為大家介紹了python多線程性能測試之快速mock數(shù)據(jù)，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

背景

在我們測試工作中，性能測試也是避免不了的，因此在性能測試前期準(zhǔn)備工作中，需要 mock 足夠批量的數(shù)據(jù)進(jìn)行壓測。那么怎么能在短時(shí)間內(nèi)快速 mock 出想要的格式數(shù)據(jù)和足夠量的數(shù)據(jù)進(jìn)行壓測？那么往下看。

安裝相關(guān)類包

pip install kafka
pip install appmetrics
pip install faker
pip install pykafka

快速 mock kafka 批量測試數(shù)據(jù)

# -* 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():
    #數(shù)據(jù)格式可自行定義
    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  關(guān)閉同步，使用異步
    while True:
        data_uni = data_result()
        producer.produce(bytes(data_uni, encoding='utf-8'))
        meter.notify(1) # 請求一次 記錄器打點(diǎn)一次
        # 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:
            #該線程是為了記錄每秒打點(diǎn)作用
            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，度量一系列事件發(fā)生的速率 (rate)，例如 TPS。Meters 會(huì)統(tǒng)計(jì)最近 1 分鐘，5 分鐘，15 分鐘，還有全部時(shí)間的速率。

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

返回結(jié)果

{'count': 5, 'kind': 'meter', 'five': 0.0066114184713530035, 'mean': 0.27743058841197027, 'fifteen': 0.0022160607980413085, 'day': 2.3147478365093123e-05, 'one': 0.031982234148270686}

以上就是python批量測試多線程之快速mock數(shù)據(jù)的詳細(xì)內(nèi)容，更多關(guān)于python測試多線程mock數(shù)據(jù)的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: