最近接触了性能压测的一款工具Locust，分享下：
一、首先说下压测工具对比：
Jmeter：
- 开源免费：JMeter是一款免费的开源软件，使用它不需要支付任何费用
- 跨平台：java开发的开源软件
- 小巧：相比LR的庞大(LoadRunner 4GB左右)，它非常小巧
- 免安装：但需要JDK环境，因为它是使用java开发的工具
- JMeter 可以做web程序的功能测试，利用JMeter 中的样本，可以做灰盒测试
- 功能强大：jmeter设计之初只是一个简单的web性能测试工具，但经过不段的更新扩展，现在可以完成数据库、FTP、LDAP、WebService等方面的测试
- 灵活扩展：因其开源，可获取源代码进行二次开发、封装、优化，对其功能进行客制化，使其更好的适应测试需求；也可以根据自己的需求扩展它的功能，可自行编写扩展包(jar)，放在{apache-jmeter-2.12\lib\ext}目录下，通过 Java请求 引用即可
LoadRunner：
- 界面不美观(开源典型的特点)
- 结果数据展示当前而言是所有性能测试中最为全面详细的一个工具
- 录制功能、调试环境比较实用
- 有 IP 欺骗功能，ＩＰ欺骗是指在一ＰＣ台上多个ＩＰ地址来分配给并发用户。这个功能对于模拟较真实的客户环境来说，比较有用
- 商用性能测试软件，有专业技术支持，即LoadRunner主要用于性能测试
Locust：
Locust 同样是开元性能测试工具，虽然官方这样来描述它 “An open source load testing tool.” 。但其它和前面两个工具有着较大的不同。相比前面两个工具，功能上要差上不少，但它也并非优点全无。
- Locust 完全基本 Python 编程语言，采用 Pure Python 描述测试脚本，并且 HTTP 请求完全基于 Requests 库。除了 HTTP/HTTPS 协议，Locust 也可以测试其它协议的系统，只需要采用Python调用对应的库进行请求描述即可。
- LoadRunner 和 Jmeter 这类采用进程和线程的测试工具，都很难在单机上模拟出较高的并发压力。Locust 的并发机制摒弃了进程和线程，采用协程（gevent）的机制。协程避免了系统级资源调度，由此可以大幅提高单机的并发能力。
- 与Jmeter一样，支持分布式性能测试。由于单机并发量高，因此可以做数百万量级别的并发测试。
二、Locust安装：
1、安装Python：
安装Python2 或Python3
2、安装Locuse
2.1， 通过pip命令安装 ，终端输入：pip install locustio
3、安装 pyzmq
如果你打算运行Locust 分布在多个进程/机器，我们建议你也安装pyzmq.
通过pip命令安装。 终端输入：pip install pyzmq
4、安装成功，终端输入locust --help，验证locust安装是否完成。
三、locust使用：
项目实例（见者勿喷，代码多少有点Bug，未做优化）
from collections import Mappingimport queuefrom web3 import Web3, HTTPProvider# from locust import HttpLocust,TaskSequence,seq_taskfrom locust import HttpLocust,TaskSet,taskimport jsonimport timefrom web3._utils.encoding import ( remove_0x_prefix, to_bytes, to_hex)from eth_utils import keccak as eth_utils_keccak
from eth_keys import ( keys)import requests
maskBit = 4# maskBit = 0# listA = [4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19]listA = [20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35]# listB = [2]
# chainId = "2"
def signTransaction(transaction_dict, private_key): FULL_NODE_HOSTS = ‘http://192.168.1.126:8089‘
provider = HTTPProvider (FULL_NODE_HOSTS) web3 = Web3 (provider) if not isinstance(transaction_dict, Mapping): raise TypeError("transaction_dict must be dict-like, got %r" % transaction_dict) sign_str = transaction_dict["chainId"] + remove_0x_prefix(transaction_dict["from"].lower()) + \ remove_0x_prefix(transaction_dict["to"].lower()) + transaction_dict["nonce"] + \ transaction_dict["value"] + remove_0x_prefix(transaction_dict["input"].lower()) sign_bytes = to_bytes(text=sign_str) res = eth_utils_keccak(sign_bytes) sign_hash = web3.eth.account.signHash(to_hex(res), private_key=private_key)
transaction_dict["sig"] = to_hex(sign_hash.signature) pk = keys.PrivateKey(private_key) transaction_dict["pub"] = "0x04" + pk.public_key.to_hex()[2:] return transaction_dict
# def get_privatekey():# FULL_NODE_HOSTS = ‘http://192.168.1.126:8089‘## provider = HTTPProvider (FULL_NODE_HOSTS)# web3 = Web3 (provider)# key = ‘0x15d115381a4e445d66c59f4c2b884d78a34ac54bccc333b4508bce9cacf32539‘# ret = web3.eth.account.encrypt(key, "123456")# # # 打开一个文件# keyfile = open("./keystore/key1", "w")# keyfile.write(json.dumps(ret))# ## # # 关闭打开的文件# keyfile.close()## with open("./keystore/key1") as keyfile:# encrypted_key = keyfile.read()# encrypted_keyobj = json.loads(encrypted_key)# private_key = web3.eth.account.decrypt(encrypted_keyobj, ‘123456‘)# return private_key
def GetfromAddress(): try: file = open("./fromaddress.txt", ‘r‘, encoding=‘utf-8‘) except IOError: error = [] return error fromaddresses = [] for line in file: fromaddresses.append(line.strip()) file.close() return fromaddresses
def GettoAddress(): try: file = open ("./toaddress.txt", ‘r‘, encoding=‘utf-8‘) except IOError: error = [] return error toaddresses = [] for line in file: toaddresses.append (line.strip()) file.close() return toaddresses
def GetchainId(fromaddress): addressbyte = bytes.fromhex (fromaddress[2:]) byteSize = (maskBit >> 3) +1 byteNum = addressbyte[0:byteSize] idx = ord(byteNum) mask = maskBit & 0x7 if mask == 0: return idx bits = 8 - mask idx >>= bits chainId = listA[idx] return chainId
def GetAccount(chainId,fromaddress): url = ‘http://172.26.65.237‘ headers = {‘Content-Type‘: ‘application/json‘} data = { "method": "GetAccount", "params": {"chainId": chainId, "address": fromaddress} } response = requests.post (url=url, headers=headers, data=json.dumps (data).encode (encoding=‘UTF8‘)) # print(response) # time.sleep (5) assert response.status_code if ‘error‘ in response: return response[‘error‘] resp = json.loads(response.content.decode()) # print(resp) nonceid = resp["nonce"] # print(nonceid) return nonceid
class UserBehavior(TaskSet): @task(1) def TestTransfer(self): """转账交易""" starttime = time.time() try: fromaddress = self.locust.fromaddress_queue.get() # 获取fromaddress队列里的数据，并赋值给fromaddress # print (fromaddress) except queue.Empty: # 队列取空后，直接退出 print("no data exist") exit(0) chainId = GetchainId(fromaddress) # print(chainId) #nonceid初始化，首次通过getaccount获取 # print(fromaddress) nonceid = GetAccount(str(chainId),fromaddress) # print(nonceid) # for toaddress in toaddresses: for i in range(10000): toaddress = self.locust.toaddress_queue.get() print(u‘当前转出地址：‘,fromaddress) print(u‘当前转入地址：‘,toaddress) url = ‘http://172.26.65.237‘ headers = {‘Content-Type‘: ‘application/json‘} con_tx = { "chainId": str(chainId), "fromChainId": str(chainId), "toChainId": "2", "from": fromaddress, "nonce": str(nonceid), "to": toaddress, "input": ‘‘, "value": "3" } # privartekey = get_privatekey () con_signtx = signTransaction(con_tx, b‘\x15\xd1\x158\x1aND]f\xc5\x9fL+\x88Mx\xa3J\xc5K\xcc\xc33\xb4P\x8b\xce\x9c\xac\xf3%9‘) # print (con_signtx) data = {"method": "SendTx","params":con_signtx} with self.client.post(url=url, headers=headers,data=json.dumps (data).encode (encoding=‘UTF8‘)) as response: # 设置断言（1、状态码断言；2、返回结果断言） if response.status_code != 200: # print (u"返回异常！") print (u"请求返回状态码:", response.status_code) elif response.status_code == 200: # print (u"返回正常！") if ‘TXhash‘ in json.loads (response.content.decode ()): print (u‘交易请求发送成功！‘) else: print (u‘请求结果为空，请确认请求参数是否正确！‘) # 每个账户每次执行请求后，nonce值加1，做循环请求 nonceid = nonceid + 1 print(time.time()-starttime) # resp = json.loads (response.content.decode ()) # # 提取交易请求返回的TXhash值 # TXhash = resp["TXhash"] # # print (nonceid) # return TXhash
class websitUser(HttpLocust): task_set = UserBehavior #从文本中读取fromaddress地址，并加入队列 fromaddresses = GetfromAddress () fromaddress_queue = queue.Queue() for fromaddress in fromaddresses: fromaddress_queue.put_nowait(fromaddress) toaddresses = GettoAddress () toaddress_queue = queue.Queue () for toaddress in toaddresses: toaddress_queue.put_nowait(toaddress)
min_wait = 10 # 单位毫秒 max_wait = 2000 # 单位毫秒
说明：红色标记部分为Locust自带的参数，具体参数说明可网上查找。另在import中，红色标记注释的那一段TaskSequence,seq_task可用来处理流程类的任务，即按照标记的先后顺序执行。
- Locust类：
用法：类名(TaskSet)
每生成一个实例都代表一个虚拟的用户，用来发送请求到进行负载测试的系统。
该用户的行为由task_set属性定义，该属性应指向一个 TaskSet类。
这个类通常应该由某些类继承并且重新定义。例如，当测试HTTP系统时，使用的HttpLocust类。
max_wait = 1000
执行locust任务之间的最长等待时间,单位是毫秒
min_wait = 1000
执行locust任务之间的最短等待时间，单位是毫秒
task_set =TaskSet
指向TaskSet类，定义了locust的执行行为
weight = 10
一个测试用例中添加多个locust实例，每个locust实例执行占的比重，数字越大，调用的频率越高。（一般用法为@task()）
- HttpLocust类
用法：类名(HttpLocust)
继承了Locust类，表示将要生成的每一个虚拟的HTTP用户，用来发送请求到进行负载测试的系统。
该用户的行为由task_set属性定义，该属性应指向一个 TaskSet类。
此类在实例化时比Locust会多了一个client属性，该属性是支持在请求之间保留用户session。
client =无
在Locust实例化时创建的HttpSession实例。client支持cookie，可以保持HTTP请求之间的session。
原文地址：https://www.cnblogs.com/Calainkey/p/11063628.html
时间： 2024-08-01 20:10:58