今天小编给大家分享一下python多线程性能测试之快速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数据测试分析”这篇文章的所有内容，感谢各位的阅读！相信大家阅读完这篇文章都有很大的收获，小编每天都会为大家更新不同的知识，如果还想学习更多的知识，请关注天达云行业资讯频道。