标签：with open app header -- efi 开源 录制 elf session
最近接触了性能压测的一款工具Locust，分享下：
一、首先说下压测工具对比：
Jmeter：
LoadRunner：
Locust：
Locust 同样是开元性能测试工具，虽然官方这样来描述它 “An open source load testing tool.” 。但其它和前面两个工具有着较大的不同。相比前面两个工具，功能上要差上不少，但它也并非优点全无。
Locust 完全基本 Python 编程语言，采用 Pure Python 描述测试脚本，并且 HTTP 请求完全基于 Requests 库。除了 HTTP/HTTPS 协议，Locust 也可以测试其它协议的系统，只需要采用Python调用对应的库进行请求描述即可。
LoadRunner 和 Jmeter 这类采用进程和线程的测试工具，都很难在单机上模拟出较高的并发压力。Locust 的并发机制摒弃了进程和线程，采用协程（gevent）的机制。协程避免了系统级资源调度，由此可以大幅提高单机的并发能力。
二、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‘\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可用来处理流程类的任务，即按照标记的先后顺序执行。
用法：类名(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)
继承了Locust类，表示将要生成的每一个虚拟的HTTP用户，用来发送请求到进行负载测试的系统。
该用户的行为由task_set属性定义，该属性应指向一个 TaskSet类。
此类在实例化时比Locust会多了一个client属性，该属性是支持在请求之间保留用户session。
client =无
在Locust实例化时创建的HttpSession实例。client支持cookie，可以保持HTTP请求之间的session。
标签：with open app header -- efi 开源 录制 elf session
原文地址：https://www.cnblogs.com/Calainkey/p/11063628.html
