第 7 章 · TaskSet 与嵌套任务
本章目标：掌握 TaskSet 类的编写方式，实现任务的分层组织与嵌套调度，学会用
interrupt()跳出子任务集，理解on_start/on_stop生命周期方法。
7.1 为什么需要 TaskSet
当用户行为模型变得复杂时（如"先登录 → 浏览商品 → 加入购物车 → 结算"），把所有 @task 堆在一个 HttpUser 类里会让代码难以维护。TaskSet 允许你将任务按业务场景分组：
from locust import HttpUser, TaskSet, task, between
class BrowsingBehavior(TaskSet):
"""模拟浏览商品的用户行为"""
@task(5)
def view_products(self):
self.client.get("/api/products")
@task(3)
def view_product_detail(self):
self.client.get("/api/products/1")
@task(1)
def stop_browsing(self):
# 离开浏览模式
self.interrupt()
class ShoppingUser(HttpUser):
wait_time = between(1, 3)
host = "https://shop.example.com"
tasks = {BrowsingBehavior: 10} # 权重为 10HttpUser 通过 tasks 属性引用一个或多个 TaskSet。当 TaskSet 被选中后，其中的任务开始循环执行，直到调用 interrupt() 返回上层。
7.2 嵌套 TaskSet
TaskSet 可以多层嵌套，构建复杂的用户旅程模型：
class LoginTask(TaskSet):
@task
def login(self):
resp = self.client.post("/api/login", json={
"username": "test_user",
"password": "secret"
})
if resp.status_code == 200:
# 登录成功，进入购物流程
self.interrupt() # 返回父级
class ShoppingTask(TaskSet):
tasks = {
BrowseProducts: 5,
AddToCart: 3,
Checkout: 1
}
@task
def stay_here(self):
pass # 保持当前层级的兜底任务
class EcommerceUser(HttpUser):
wait_time = between(1, 3)
host = "https://shop.example.com"
tasks = [LoginTask]每层 TaskSet 都有自己的任务权重系统。interrupt(reschedule=True) 可以在跳出后立即重新选择下一个任务。
7.3 interrupt() 的用法
interrupt() 是从子 TaskSet 返回到父级的关键方法。不调用它的话，agent 将永远留在当前 TaskSet 中循环。
class SessionTimeout(TaskSet):
@task
def check_session(self):
resp = self.client.get("/api/session/check")
if resp.json().get("expired"):
self.interrupt(reschedule=True) # 会话过期 → 返回上级重新登录
@task
def do_work(self):
self.client.get("/api/work")# interrupt() 参数说明
self.interrupt() # 跳出后等待下一轮再选任务
self.interrupt(reschedule=True) # 跳出后立即重新调度忘记 interrupt()
最常见的错误是在子 TaskSet 中忘记设置退出条件，导致所有虚拟用户永远卡在该层级。确保每个子 TaskSet 都有至少一个能触发 interrupt() 的路径。
7.4 on_start 和 on_stop 生命周期
每个 TaskSet（和 User）都支持两个生命周期方法：
- on_start：虚拟用户启动时执行一次（如登录、获取 token）
- on_stop：虚拟用户停止时执行一次（如登出、清理资源）
class AuthenticatedUser(HttpUser):
host = "https://api.example.com"
wait_time = between(1, 3)
def on_start(self):
"""每个虚拟用户启动时执行一次"""
resp = self.client.post("/api/auth/login", json={
"username": f"user_{id(self)}",
"password": "test123"
})
self.token = resp.json()["token"]
self.headers = {"Authorization": f"Bearer {self.token}"}
def on_stop(self):
"""虚拟用户停止时执行清理"""
self.client.post(
"/api/auth/logout",
headers=self.headers
)
class BrowseTask(TaskSet):
@task
def get_data(self):
# 使用父类的 token
self.client.get(
"/api/data",
headers=self.parent.headers # 访问父级属性
)访问父级
在嵌套 TaskSet 内部通过 self.parent 访问父级 User 或 TaskSet 实例的属性和方法。
7.5 SequentialTaskSet：顺序执行
默认情况下 TaskSet 中的任务按权重随机选取。如果你需要严格按顺序执行（如注册→验证→完成），使用 SequentialTaskSet：
from locust import SequentialTaskSet, HttpUser, task
class RegistrationFlow(SequentialTaskSet):
"""严格按照顺序执行的注册流程"""
@task
def step1_submit_form(self):
self.client.post("/api/register", json={"email": "new@test.com"})
# 保存返回值供下一步使用
self.user_id = self.client.get("/api/me").json()["id"]
@task
def step2_verify_email(self):
self.client.post("/api/verify", json={"user_id": self.user_id})
@task
def step3_complete_profile(self):
self.client.put(f"/api/users/{self.user_id}", json={
"nickname": "新用户"
})
class SignupUser(HttpUser):
host = "https://app.example.com"
tasks = [RegistrationFlow]SequentialTaskSet 执行完最后一步后会自动回到第一步继续循环，除非你调用 interrupt()。
本章小结
TaskSet按业务场景分组任务，通过父类tasks属性引用；- 多层嵌套构建复杂用户旅程，
interrupt()跳出子层级； on_start/on_stop处理一次性初始化和清理；SequentialTaskSet保证任务按声明顺序执行；- 子 TaskSet 通过
self.parent访问父级属性。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 从子 TaskSet 返回到父级的方法是？
2. on_start 方法在什么时候被调用？
3. SequentialTaskSet 与普通 TaskSet 的核心区别是？
4. 在嵌套 TaskSet 中如何访问顶层 User 的属性？
🛠️ 动手实践
- 设计三层嵌套 TaskSet：登录 → 浏览 → 下单，每层有独立的权重配置。
- 在子 TaskSet 中使用
interrupt()实现"会话超时自动重新登录"逻辑。 - 用
SequentialTaskSet编写完整的"密码找回"压测流程（请求重置→验证码校验→设置新密码）。