我需要对一个系统进行压力测试,http://locust.io看起来是这样做的最佳方式。然而,它似乎设置为每次使用相同的用户。我需要每个生成的用户登录不同的用户。我该如何设置?或者,是否有其他可以使用的系统?
我需要对一个系统进行压力测试,http://locust.io看起来是这样做的最佳方式。然而,它似乎设置为每次使用相同的用户。我需要每个生成的用户登录不同的用户。我该如何设置?或者,是否有其他可以使用的系统?
Locust的作者在这里。
默认情况下,每个HttpLocust用户实例都有一个HTTP客户端,它有它自己独立的会话。
Locust没有提供提供用户凭据列表或类似功能的特性。但是,您的负载测试脚本只是Python代码,幸运的是,自己实现这个功能非常简单。
这里是一个简短的示例:
# locustfile.py
from locust import HttpLocust, TaskSet, task
USER_CREDENTIALS = [
("user1", "password"),
("user2", "password"),
("user3", "password"),
]
class UserBehaviour(TaskSet):
def on_start(self):
if len(USER_CREDENTIALS) > 0:
user, passw = USER_CREDENTIALS.pop()
self.client.post("/login", {"username":user, "password":passw})
@task
def some_task(self):
# user should be logged in here (unless the USER_CREDENTIALS ran out)
self.client.get("/protected/resource")
class User(HttpLocust):
task_set = UserBehaviour
min_wait = 5000
max_wait = 60000
以上代码在运行Locust分布式时无法工作,因为相同的代码在每个从节点上运行,它们不共享任何状态。因此,您需要引入一些外部数据存储库,供从节点使用以共享状态(例如PostgreSQL、redis、memcached或其他)。
或者,您可以创建users.py
模块来保存测试用例中所需的用户信息,在我的示例中,它保存了电子邮件
和cookies
。然后您可以在任务中随机调用它们。请参见以下内容:
# locustfile.py
from locust import HttpLocust, TaskSet, task
from user_agent import *
from users import users_info
class UserBehaviour(TaskSet):
def get_user(self):
user = random.choice(users_info)
return user
@task(10)
def get_siparislerim(self):
user = self.get_user()
user_agent = self.get_user_agent()
r = self.client.get("/orders", headers = {"Cookie": user[1], 'User-Agent': user_agent})
class User(HttpLocust):
task_set = UserBehaviour
min_wait = 5000
max_wait = 60000
用户和用户代理可以通过函数调用。这种方式,我们可以使用许多用户和不同的用户代理来分发测试。
# users.py
users_info = [
['performancetest.1441926507@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926506@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926501@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926499@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926494@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926493@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926492@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926491@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926490@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926489@gmail.com', 'cookies_created_by_each_user'],
['performancetest.1441926487@gmail.com', 'cookies_created_by_each_user']]
USER_CREDENTIALS
列表,并开始抛出错误。USER_CREDENTIALS = generate_users()
@events.spawning_complete.add_listener
def spawn_complete_handler(**kw):
global USER_CREDENTIALS
USER_CREDENTIALS = generate_users()
from locust import events # in addition to the other locust modules needed
def hatch_complete_handler(**kw):
global USER_CREDENTIALS
USER_CREDENTIALS = generate_users() # some function here to regenerate your list
events.hatch_complete += hatch_complete_handler
当你的群集孵化完成后,这将刷新您的用户列表。
还要记住,您需要一个比您希望生成的用户数量更长的列表。
USER_LIST = generate_users()
@events.spawning_complete.add_listener
def spawn_complete_handler(**kw):
global USER_LIST
USER_LIST = generate_users()
- Gh0sT当我在实现这个分布式系统时,我采用了稍微不同的方法。在TaskSet的on_start部分中,我使用了一个非常简单的Flask服务器,通过get调用它。
from flask import Flask, jsonify
app = Flask(__name__)
count = 0 #Shared Variable
@app.route("/")
def counter():
global count
count = count+1
tenant = count // 5 + 1
user = count % 5 + 1
return jsonify({'count':count,'tenant':"load_tenant_{}".format(str(tenant)),'admin':"admin",'user':"load_user_{}".format(str(user))})
if __name__ == "__main__":
app.run()
这样一来,我现在有了一个端点,无论在哪个主机上运行,我都可以访问http://localhost:5000/。我只需要使这个端点对从系统可访问,就不必担心重复用户或由于具有有限的user_info集而导致的某种轮询效应。