性能测试 - Locust dubbo client
Locust dubbo客户端实现# _*_ coding:utf-8 _*_
'''
Locust dubbo client
Max.Bai
2018-08
'''
import time
import random
import json
import socket
import dubbo_telnet
import requests
from locust import Locust, TaskSet, events, task
class DubboClient(object):
def __init__(self, host, port):
self.conn = dubbo_telnet.connect(host, port)
self.conn.set_connect_timeout(10)
def do(self,msg):
start_time = time.time()
try:
self.conn.do(msg)
except Exception as e:
total_time = int((time.time() - start_time) * 1000)
events.request_failure.fire(request_type="dubbo", name="start", response_time=total_time, exception=e)
else:
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(request_type="dubbo", name="start", response_time=total_time, response_length=0)
class DubboLocust(Locust):
"""
This is the abstract Locust class which should be subclassed. It provides an dubbo client
that can be used to make dubbo requests that will be tracked in Locust's statistics.
"""
def __init__(self, *args, **kwargs):
super(DubboLocust, self).__init__(*args, **kwargs)
self.client = DubboClient(self.host, self.port)
class TestUser(DubboLocust):
host = "119.29.67.113"
port = 20904
min_wait = 2000
max_wait = 2000
class task_set(TaskSet):
@task
def send_data1(self):
msg = 'invoke com.abc.com.base.service.faceattendance.abcService.action("parm1", "parm2", "parm3")'
response = self.client.do(msg)
print "recv:%s"%str(response)
if __name__ == "__main__":
user = TestUser()
user.run()
感谢分享 感谢分享
页:
[1]
