性能测试之Locust(一)

最近接触了性能压测的一款工具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 欺骗功能,IP欺骗是指在一PC台上多个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:

官方:https://www.python.org/

安装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-10-01 16:28:05

性能测试之Locust(一)的相关文章

【原创】性能测试之——网络环境分析

性能测试之——网络环境分析 首先,我们需要了解宽带上网时的网络带宽环境概念: 这里指的是带宽网速的单位计算方式方法及关系. 在计算机网络.IDC机房中,其宽带速率的单位用bps(或b/s)表示:换算关系为:1Byte=8bit 1B=8b             ---------- 1B/s=8b/s(或1Bps=8bps) 1KB=1024B     ---------- 1KB/s=1024B/s 1MB=1024KB  ---------- 1MB/s=1024KB/s 在实际上网应用中

【原创】性能测试之——性能测试需求分析

性能测试之——性能测试需求分析 这里以一个电商购物(B2C)网站为例: 客户的购物网站性能测试(业务)需求: 从12月下旬至农历年底(来年2月初)(<=50天)网站预计营业额(400万),这里营业额可以理解为网站完成购买订单总金额: 访问订单转化率:10%,这里理解为百分之多少的访问量会转化为实际的网站订单: 每日访问时间:24小时×80%,这里理解为正常用户会在早6点至凌晨0点之前进行电子购物,下午18点下班至晚上22点为购物高峰期: 每个订单平均选购商品数:3件左右共计300元左右的金额,这

[Android Pro] Android应用性能测试之CPU和内存占用(转载)

首先稍做分析一下测试环境:我们知道CPU和内存占用是一个实时变化的状态,而市面上还没有具体的哪款android应用能做到实时监控CPU和内存占用并使用log日志保存.考虑到android的底层框架是基于Linux的平台,所有我们可以通过Linux的资源监控命令来实现对android平台的资源实时监控. 要做到上边的测试环境的实现,需要具备以下几点: 1.被测试的手机具备root权限:因为涉及到底层的linux命令,需要读取或执行相应的文件.至于如何root你的手机,不同型号的手机root的方法不

性能测试之-wrk(转)

性能测试之-wrk(转) 转载地址:http://zjumty.iteye.com/blog/2221040 http://www.cnblogs.com/rainy-shurun/p/5867946.html 测试先行是软件系统质量保证的有效手段. 在单元测试方面, 我们有非常成熟的 xUnit 方案. 在集成测试方面, 我们 selenium 等自动化方案. 在性能测试方面也有很多成熟的工具, 比如 LoadRunner, Jmeter 等. 但是很多工具都是给专门的性能测试人员使用的, 功

老李分享知识:性能测试之TPS和吞吐率

老李分享知识:性能测试之TPS和吞吐率      当增大系统的压力(或添加并发用户数)时,吞吐率和TPS的改变曲线呈大体一致,则系统基本稳定. 若压力增大时,吞吐率的曲线添加到一定程度后出现改变缓慢,甚至平坦,同时TPS也趋于平坦,查看系统资源运用,假如资源运用率比较低,说明服务器硬件资源不存在疑问,查看网络流量,估计网络带宽存在疑问. 同理若点击率/TPS曲线出现改变缓慢或者平坦, 点击率(用户每秒发出的请求数)假如在压力添加时,趋于平坦,很可能是服务器响应时间添加,观察服务器资源运用情况,确

性能测试培训:Ajax接口级性能测试之jmeter版

性能测试培训:Ajax接口级性能测试之jmeter版 poptest是国内唯一一家培养测试开发工程师的培训机构,以学员能胜任自动化测试,性能测试,测试工具开发等工作为目标.在poptest认为工具不是问题,jmeter还是loadrunner都是工具级别的,真的提高性能测试水平,还是需要具备一定架构知识,网络知识,服务器方面的知识,poptest通过大量的实战案例的讲解提高学员的实战经验,尽快上手性能测试.(大家对课程感兴趣,请加qq:908821478) 1.  被测程序环境部署 对于自动化测

性能测试之Windows常见性能计数器

性能计数器(counter)是描述服务器或操作系统性能的一些数据指标.计数器在性能测试中发挥着“监控和分析”的关键作用,尤其是在分析系统的可扩展性.进行性能瓶颈的定位时,对计数器的取值的分析非常关键.但必须说明的是,单一的性能计数器只能体现系统性能的某一个方面,对性能测试结果的分析必须基于多个不同的计数器. 与性能计数器相关的另一个术语是“资源利用率”.该术语指的是系统各种资源的使用状况.为了方便比较,一般用“资源的实际使用/总的资源可用量”形成资源利用率的数据,用以进行各种资源使用的比较. 性

FTP服务器 传输性能测试之Raid 1+0篇

FTP服务器  传输性能测试之Raid 1+0篇 2012年02月09日13:27 it168网站原创 作者:于泽 编辑:于泽 查看全文 赞(0)评论(1) 分享 [IT168 评测]作为日常办公最常用到的一种应用服务器,FTP服务器承担着很多工作任务,而在FTP服务器的各项性能指标中,传输效率无疑是人们关注的首要因素,在FTP服务器该做Raid 5吗?传输性能评测一文中,我们曾介绍过在Raid 5模式下,FTP服务器的传输表现.今天我们将在同样的平台下,分别对服务器做Raid 1+0.Raid

网页性能测试之WebPageTest

想知道您的网站,性能怎么样? 很自然,首先得找一个广被认可的测试工具.我们推荐WebPageTest: WebPageTest 它是google 开源项目"make the web faster "的子项目("make the web faster包括page speed,spdy,tcpm等等"),它本来是AOL内部使用的工具,后来在2008年基于BSD开源.其网址如下:http://www.webpagetest.org/ 我们抓取出来里面的主要指标. 那么,问