locust扩展支持其它协议压测

语言版本：python2.7 

一：dnsClient.py

# -*- coding: utf-8 -*-

# @Time    : 2020-01-08 15:32

# @Author  : xiaobin

# @File    : dnsClient.py

# @Software : PyCharm

import time

from locustimport (TaskSet,task,events,Locust)

from gevent._semaphoreimport Semaphore

import socket

all_locusts_spawned = Semaphore()

all_locusts_spawned.acquire()

def on_hatch_complete(**kwargs):

all_locusts_spawned.release()

events.hatch_complete += on_hatch_complete

class dnsClient(object):

def look_host(self, hostname):

start_time = time.time()

time.sleep(0.195)

resp =self.get_host(hostname)

total_time =int((time.time() - start_time) *1000)

print total_time

if resp:

events.request_success.fire(

request_type='dns-test',

                name=r'dns',

                response_time=total_time,

                response_length=0)

else:

events.request_success.fire(

request_type ='dns-test',

            name =r'dns-error',

            response_time = total_time,

            exception ='error')

def get_host(self, hostname):

falg =False

        try:

result = socket.gethostbyaddr(hostname)

falg =True

        except socket.herror, e:

print e

return falg

if __name__ =='__main__':

d = dnsClient()

d.look_host('dnstest.com')

二：locustDns.py

# -*- coding: utf-8 -*-

# @Time    : 2020-01-08 15:47

# @Author  : xiaobin

# @File    : locustDns.py

# @Software : PyCharm

from locustimport (TaskSet,task,events,Locust)

import core.dns_test.dnsClient

class locustDns(Locust):

def __init__(self):

super(locustDns, self).__init__()

self.client = core.dns_test.dnsClient.dnsClient()

三：LocustRun.py

# -*- coding: utf-8 -*-

# @Time    : 2020-01-08 15:52

# @Author  : xiaobin

# @File    : TestDns.py

# @Software : PyCharm

import sys

import os

curPath = os.path.abspath(os.path.dirname(__file__))

rootPath = os.path.split(curPath)[0]

sys.path.append(rootPath)

from locustimport *

import core.dns_test.locustDns

class LocustRun(TaskSet):

def setup(self):

print('task setup')

def teardown(self):

print('task teardown')

def on_start(self):

print('start')

def on_stop(self):

# 虚拟用户结束Task时运行

        print('end')

@task(weight=1)

def test_dns(self):

self.client.look_host('dnstest.com')

class myrun(core.dns_test.locustDns.locustDns):

task_set = LocustRun

host ='http://127.0.0.1:8888'

    min_wait =0

    max_wait =0

四：运行locust -f LocustRun.py 就可以了

     记得本地/etc/hosts 增加测试域名