第python多線程性能測(cè)試之快速mock數(shù)據(jù)目錄背景安裝相關(guān)類包快速mockkafka批量測(cè)試數(shù)據(jù)appmetrics使用方法Meters

背景

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

安裝相關(guān)類包

pipinstallkafkapipinstallappmetricspipinstallfakerpipinstallpykafka

快速mockkafka批量測(cè)試數(shù)據(jù)

#-*coding:utf8*-

frompykafkaimportKafkaClient

importuuid

importtime

importthreading

fromappmetricsimportmetrics

fromfakerimportFaker

importos

fake=Faker("zh-cn")

PATH=lambdap:os.path.abspath(

os.path.join(os.path.dirname(__file__),p)

meter=metrics.new_meter("meter_test")

host_producer='host地址'

defdata_info():

uid=str(uuid.uuid4())

suid=''.join(uid.split('-'))

returnsuid

defdata_result():

#數(shù)據(jù)格式可自行定義

data=f"{data_info()},{fake.phone_number()},111111111111,LOL-UZI"

returndata

defmock_request():

client_producer=KafkaClient(hosts=host_producer)

topicdocu=client_producer.topics['XXXXXXX-TOPIC']

producer=topicdocu.get_producer(sync=False)#sync=False關(guān)閉同步，使用異步

whileTrue:

data_uni=data_result()

duce(bytes(data_uni,encoding='utf-8'))

meter.notify(1)#請(qǐng)求一次記錄器打點(diǎn)一次

#i=i-1

producer.stop()

defprint_meter():

whileTrue:

print(meter.get())

time.sleep(1)

defthread_request(nums):

t1=[]

foriinrange(nums):

ifi==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)

fortint1:

t.start()

fortint1:

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.314747836509312