• Locust 之 TaskSet类梳理

            TaskSet类翻译过来就是任务集合，存放的是locust中user用户要执行的任务集合。来看看这个类的构造。

    一、成员变量

            1.1、 tasks 任务集合，存放任务

    1. tasks: List[Union["TaskSet", Callable]] = []
    2. """
    3. Collection of python callables and/or TaskSet classes that the User(s) will run.
    5. If tasks is a list, the task to be performed will be picked randomly.
    7. If tasks is a *(callable,int)* list of two-tuples, or a {callable:int} dict,
    8. the task to be performed will be picked randomly, but each task will be weighted
    9. according to its corresponding int value. So in the following case, *ThreadPage* will
    10. be fifteen times more likely to be picked than *write_post*::
    12.     class ForumPage(TaskSet):
    13.         tasks = {ThreadPage:15, write_post:1}
    14. """

            task是一个List类型的对象，里面存放的对象类型为TaskSet或Callable，这些都是即将在User对象中被运行的任务。所以一个TaskSet任务集合对象中可以嵌套另外一个任务集合对象。

            Callable代表一个可执行类型，即可以在该对象后面直接加上()执行！

            注释中说明了，如果tasks是一个list列表，则列表中任务将被随机执行；

            如果tasks是一个存放数据类型为(callable,int)二元组的列表，或{callable:int}字典类型，则任务task将被随机执行，但是每个任务的权重依赖相应的int值。所以，像下面这种情况ThreadPage任务执行的可能性大概是write_post任务的15倍。

    1. class ForumPage(TaskSet):
    2.      tasks = {ThreadPage:15, write_post:1}

    1.2、 _user 模拟用户对象  _parent模拟用户对象的父亲对象

            _user和_parent是类变量。

    1. _user: "User"
    2. _parent: "User"

            TaskSet对象是作为User模拟用户对象的一个成员变量的角色。在TaskSet中绑定了是哪个User对象来执行 TaskSet对象中任务集合tasks中的任务。

    1.3、构造函数 __init__(self, parent: "User")

    1. def __init__(self, parent: "User") -> None:
    2.     self._task_queue: List[Callable] = []
    3.     self._time_start = time()
    4.  
    5.     if isinstance(parent, TaskSet):
    6.         self._user = parent.user
    7.     else:
    8.         self._user = parent
    9.  
    10.     self._parent = parent
    11.  
    12.     # if this class doesn't have a min_wait, max_wait or wait_function defined, copy it from Locust
    13.     if not self.min_wait:
    14.         self.min_wait = self.user.min_wait
    15.     if not self.max_wait:
    16.         self.max_wait = self.user.max_wait
    17.     if not self.wait_function:
    18.         self.wait_function = self.user.wait_function

            成员变量 self._task_queue任务队列是List[Callable]， 默认为空；

            成员变量 _time_start 为当前时间time()；

            可以看到在构造器中给TaskSet的实例对象上又定义了两个成员变量_user和_parent，注意区别类中定义的类变量_user和_parent。

            构造器参数parent期望传入的是User实例对象，但实际可能不是。构造器中会对parent参数类型进行判断，如果传进来的parent对象是TaskSet类型，则让当前TaskSet对象的_user指向parent的user。如果传进来的parent对象是User类型，则直接把parent赋值给当前TaskSet对象的_user。

            最后执行self._parent = parent。

    这里总结下：

            如果传入TaskSet构造器的parent是User对象，则TaskSet对象的_user和_parent都指向parent;

            如果传入TaskSet构造器的parent是另外一个TaskSet对象，则当前TaskSet对象的_user执行传入TaskSet对象parent的user属性对象，当前TaskSet对象的_parent指向传入的TaskSet对象parent。

            综上，

                    _user指向的是创建TaskSet任务的User实例对象；

                    _parent要分情况，parent参数是User，则_parent指向该User; parent参数是TaskSet，则_parent指向传入的parent；

                    总之，当前TaskSet的父亲_parent是传入的parent对象。

            这里好绕啊！！后面再分析吧

    1. if isinstance(parent, TaskSet):
    2.     self._user = parent.user
    3. else:
    4.     self._user = parent
    5.  
    6. self._parent = parent

    二、成员函数

            1、parent(self)

    1. @property
    2. def parent(self):
    3.     """Parent TaskSet instance of this TaskSet (or :py:class:`User ` if this is not a nested TaskSet)"""
    4.     return self._parent

            TaskSet如果嵌套的话，则parent指向的是外部的TaskSet；如果不嵌套的话，指向的是User对象。

            2、on_start(self)

    1. def on_start(self):
    2.     """
    3.     Called when a User starts executing this TaskSet
    4.     """
    5.     pass

            当User对象开始执行当前TaskSet对象中的任务时被调用。

            3、on_stop(self)

    1. def on_stop(self):
    2.     """
    3.     Called when a User stops executing this TaskSet. E.g. when TaskSet.interrupt() is called
    4.     or when the User is killed
    5.     """
    6.     pass

            当User对象结束任务时被调用。结束执行的情况包括调用了TaskSet.interrupt()或User对象被杀死了。

            4、schedule_task(self, task_callable, first=False)

    1. def schedule_task(self, task_callable, first=False):
    2.     """
    3.     Add a task to the User's task execution queue.
    5.     :param task_callable: User task to schedule.
    6.     :param first: Optional keyword argument. If True, the task will be put first in the queue.
    7.     """
    8.     if first:
    9.         self._task_queue.insert(0, task_callable)
    10.     else:
    11.         self._task_queue.append(task_callable)

            往任务队列_task_queue中添加任务。

            5、get_next_task(self)

    1.  def get_next_task(self):
    2.      if not self.tasks:
    3.          raise Exception(
    4.              f"No tasks defined on {self.__class__.__name__}. use the @task decorator or set the tasks property of the TaskSet"
    5.          )
    6.      return random.choice(self.tasks)

            随机取出任务队列中的一个任务。

            6、execute_task(self, task)

    1. def execute_task(self, task):
    2.     # check if the function is a method bound to the current locust, and if so, don't pass self as first argument
    3.     if hasattr(task, "__self__") and task.__self__ == self:
    4.         # task is a bound method on self
    5.         task()
    6.     elif hasattr(task, "tasks") and issubclass(task, TaskSet):
    7.         # task is another (nested) TaskSet class
    8.         task(self).run()
    9.     else:
    10.         # task is a function
    11.         task(self)

     情况1：task是绑定在当前locust上的一个函数，则不能传入self参数

            hasattr(task, "__self__")，task拥有__self__属性说明task一个实例对象的函数类型对象（这个在其他文章中证明）。

            task.__self__ == self，说明持有task变量的类实例就是当前自己本身。

            满足上面两个情况，则验证了task就是当前TaskSet类型实例对象的一个函数对象。则直接执行task()。

            对象调用自己的函数，函数不需要传self参数了。

    情况2：task是另外一个嵌套TaskSet类

            task变量具有tasks属性，并且是TaskSet的子类型，则执行task(self).run()。

            task(self)，其实调用的的是TaskSet的__init__(self, parent)构造方法，并把self传给parent参数，这样就使得当前TaskSet对象变成了新创建的TaskSet对象的父亲啦！！实在是妙啊！！！

            然后再执行新创建的TaskSet对象的run()方法，就让该TaskSet对象也运行起来了。

    其他情况：task是一个没有绑定到TaskSet类实例对象上的函数

            执行task(self)

            从上面tasks: List[Union["TaskSet", Callable]]可以知道tasks中可以存放TaskSet类实例和可调用对象。其中可调用对象表示可执行添加()运行的。可以把在函数文件中定义的函数传入tasks，不过要求该函数参数类型是TaskSet。即类似这样一个函数：

    1. def doSomeThing(taskSet: TaskSet):
    2.     pass

           所以task(self)把当前对象传入到函数中了。

            这里可以参考文章：locust官方文档-Tasks

            7、execute_next_task(self)

    1.  def execute_next_task(self):
    2.      self.execute_task(self._task_queue.pop(0))

            执行任务队列中下一个任务。

            8、run(self)

    1.     @final
    2.     def run(self):
    3.         try:
    4.             self.on_start()
    5.         except InterruptTaskSet as e:
    6.             if e.reschedule:
    7.                 raise RescheduleTaskImmediately(e.reschedule).with_traceback(e.__traceback__)
    8.             else:
    9.                 raise RescheduleTask(e.reschedule).with_traceback(e.__traceback__)
    10.  
    11.         while True:
    12.             try:
    13.                 if not self._task_queue:
    14.                     self.schedule_task(self.get_next_task())
    15.  
    16.                 try:
    17.                     if self.user._state == LOCUST_STATE_STOPPING:
    18.                         raise StopUser()
    19.                     self.execute_next_task()
    20.                 except RescheduleTaskImmediately:
    21.                     pass
    22.                 except RescheduleTask:
    23.                     self.wait()
    24.                 else:
    25.                     self.wait()
    26.             except InterruptTaskSet as e:
    27.                 self.on_stop()
    28.                 if e.reschedule:
    29.                     raise RescheduleTaskImmediately(e.reschedule) from e
    30.                 else:
    31.                     raise RescheduleTask(e.reschedule) from e
    32.             except (StopUser, GreenletExit):
    33.                 self.on_stop()
    34.                 raise
    35.             except Exception as e:
    36.                 self.user.environment.events.user_error.fire(user_instance=self, exception=e, tb=e.__traceback__)
    37.                 if self.user.environment.catch_exceptions:
    38.                     logger.error("%s\n%s", e, traceback.format_exc())
    39.                     self.wait()
    40.                 else:
    41.                     raise

            1、首先执行自身的on_start()方法，所以这里你就知道为什么on_start()会在task任务之前执行了；

            2、如果任务队列_task_queue为空了(队列中的任务被处理完了)，则self.schedule_task(self.get_next_task())负责把tasks列表中的task任务加载到_task_queue任务队列中；

            3、检测当前user状态，如果为stop状态，则抛出一个停止测试的异常；

            4、否则self.execute_next_task()执行下一个任务。

            9、client(self)

    1. @property
    2. def client(self):
    3.     """
    4.     Shortcut to the client :py:attr:`client ` attribute of this TaskSet's :py:class:`User `
    5.     """
    6.     return self.user.client

            返回user的client对象。

    最后想在补充下TaskSet的metaclass=TaskSetMeta的用法。

    class TaskSet(metaclass=TaskSetMeta):

    MetaClass 元类动态地修改内部的属性或者方法,参考文章”Python MetaClass元类详解“

    1. class TaskSetMeta(type):
    2.     """
    3.     Meta class for the main User class. It's used to allow User classes to specify task execution
    4.     ratio using an {task:int} dict, or a [(task0,int), ..., (taskN,int)] list.
    5.     """
    6.  
    7.     def __new__(mcs, classname, bases, class_dict):
    8.         class_dict["tasks"] = get_tasks_from_base_classes(bases, class_dict)
    9.         return type.__new__(mcs, classname, bases, class_dict)

            从注释可以看到，TaskSetMeta是为了让User可以指定执行权重的，通过{task：int}字典或[(task0,int), ..., (taskN,int)]这样的列表来设置不同task具有不同权重。

            我们知道在执行创建TaskSet实例时会执行__int__()方法，其实在执行__init__方法前，会先执行__new__()方法。

    1. def get_tasks_from_base_classes(bases, class_dict):
    2.     """
    3.     Function used by both TaskSetMeta and UserMeta for collecting all declared tasks
    4.     on the TaskSet/User class and all its base classes
    5.     """
    6.     new_tasks = []
    7.     for base in bases:
    8.         if hasattr(base, "tasks") and base.tasks:
    9.             new_tasks += base.tasks
    10.  
    11.     if "tasks" in class_dict and class_dict["tasks"] is not None:
    12.         tasks = class_dict["tasks"]
    13.         if isinstance(tasks, dict):
    14.             tasks = tasks.items()
    15.  
    16.         for task in tasks:
    17.             if isinstance(task, tuple):
    18.                 task, count = task
    19.                 for _ in range(count):
    20.                     new_tasks.append(task)
    21.             else:
    22.                 new_tasks.append(task)
    23.  
    24.     for item in class_dict.values():
    25.         if "locust_task_weight" in dir(item):
    26.             for i in range(item.locust_task_weight):
    27.                 new_tasks.append(item)
    28.  
    29.     return new_tasks
    30.  
    31.  
    32. class TaskSetMeta(type):
    33.     """
    34.     Meta class for the main User class. It's used to allow User classes to specify task execution
    35.     ratio using an {task:int} dict, or a [(task0,int), ..., (taskN,int)] list.
    36.     """
    37.  
    38.     def __new__(mcs, classname, bases, class_dict):
    39.         class_dict["tasks"] = get_tasks_from_base_classes(bases, class_dict)
    40.         return type.__new__(mcs, classname, bases, class_dict)
    41.  
    42.

            简而言之，通过metaclass=TaskSetMeta实现了手机当前TaskSet所有的父类的任务到tasks属性中。感觉还是没太搞清楚，以后会了再补充吧！！

  • 相关阅读:
    01 Sekiro服务器部署和第一个示例部署成功，js逆向和加解密
    关于SpringBoot项目中读取不到自建email.yml配置文件内容的问题
    TCP三次握手，四次挥手策略
    秋招我借这份PDF的复习思路，收获美团，小米，京东等Java岗offer
    python 另一种将内容写入记事本的方式
    Flutter | 让倒计时更简单
    C语言tips-字符串数组
    PyVis｜神经网络数据集的可视化
    Nvm的使用
    ARM 账号注册报错 The claims exchange ‘Salesforce-UserWriteUsingEmail‘
  • 原文地址：https://blog.csdn.net/liuqinhou/article/details/126190741