最近接触了性能压测的一款工具Locust，分享下：
一、首先说下压测工具对比：
Jmeter：
- 开源免费：JMeter是一款免费的开源软件，使用它不需要支付任何费用
- 跨平台：java开发的开源软件
- 小巧：相比LR的庞大(LoadRunner 4GB左右)，它非常小巧
- 免安装：但需要JDK环境，因为它是使用java开发的工具
- JMeter 可以做web程序的功能测试，利用JMeter 中的样本，可以做灰盒测试
- 功能强大：jmeter设计之初只是一个简单的web性能测试工具，但经过不段的更新扩展，现在可以完成数据库、FTP、LDAP、WebService等方面的测试
- 灵活扩展：因其开源，可获取源代码进行二次开发、封装、优化，对其功能进行客制化，使其更好的适应测试需求；也可以根据自己的需求扩展它的功能，可自行编写扩展包(jar)，放在{apache-jmeter-2.12libext}目录下，通过 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 Mapping
import queue
from web3 import Web3, HTTPProvider
# from locust import HttpLocust,TaskSequence,seq_task
from locust import HttpLocust,TaskSet,task
import json
import time
from 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'x15xd1x158x1aND]fxc5x9fL+x88Mxxa3Jxc5Kxccxc33xb4Px8bxcex9cxacxf3%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。