python测试开发基础---threading

本文主要是介绍python测试开发基础---threading，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1. 核心概念

线程（Thread）：线程是轻量级的进程，在同一进程内可以并行执行多个任务。线程共享进程的资源，如内存和文件描述符，但每个线程有自己的执行栈和局部变量。

全局解释器锁（GIL）：Python中的GIL限制了同一进程中多个线程的真正并行执行。它确保同一时间只有一个线程可以执行Python字节码，这对计算密集型任务可能会影响性能，但对于I/O密集型任务效果仍然良好。

2. `threading`模块的主要类和方法

2.1 `Thread` 类

功能：用来创建和管理线程。
常用方法：
- __init__(self, group=None, target=None, name=None, args=(), kwargs={})：构造函数，创建线程对象。
- start(self)：启动线程，调用run()方法。
- run(self)：线程执行的代码逻辑，子类需要重写此方法。
- join(self, timeout=None)：等待线程完成。
- is_alive(self)：检查线程是否仍在运行。

基本用法示例：

import threadingdef print_numbers():for i in range(5):print(i)# 创建线程对象
thread = threading.Thread(target=print_numbers)# 启动线程
thread.start()# 等待线程完成
thread.join()

2.2 `Lock` 类

功能：用于线程间的同步，防止数据竞争。
常用方法：
- acquire(blocking=True, timeout=-1)：获取锁，blocking指定是否阻塞，timeout指定最大等待时间。
- release()：释放锁。

使用锁的示例：

import threadinglock = threading.Lock()def thread_task():with lock:  # 自动获取和释放锁print("Thread is running")threads = [threading.Thread(target=thread_task) for _ in range(5)]for thread in threads:thread.start()for thread in threads:thread.join()

2.3 `RLock` 类

功能：可重入锁，允许同一个线程多次获取锁。
常用方法：与Lock类相同，但支持重入。

使用RLock的示例：

import threadingrlock = threading.RLock()def thread_task():with rlock:# 允许多次获取with rlock:print("Thread is running")thread = threading.Thread(target=thread_task)
thread.start()
thread.join()

2.4 `Event` 类

功能：用于线程间的简单通信，允许线程等待特定事件发生。
常用方法：
- set()：标记事件为“已发生”。
- clear()：标记事件为“未发生”。
- wait(timeout=None)：等待事件发生，阻塞直到事件标记为“已发生”或超时。

使用Event的示例：

import threading
import timeevent = threading.Event()def worker():print("Worker is waiting for event")event.wait()  # 等待事件被标记为“已发生”print("Worker has detected event")thread = threading.Thread(target=worker)
thread.start()time.sleep(2)
event.set()  # 触发事件

2.5 `Condition` 类

功能：提供线程间的复杂同步机制，允许线程在满足特定条件时继续执行。
常用方法：
- acquire()：获取锁。
- release()：释放锁。
- wait(timeout=None)：等待条件满足。
- notify(n=1)：通知一个或多个等待的线程。
- notify_all()：通知所有等待的线程。

使用Condition的示例：

import threadingcondition = threading.Condition()def consumer():with condition:print("Consumer is waiting")condition.wait()  # 等待通知print("Consumer received notification")def producer():with condition:print("Producer is notifying")condition.notify()  # 发送通知consumer_thread = threading.Thread(target=consumer)
producer_thread = threading.Thread(target=producer)consumer_thread.start()
producer_thread.start()consumer_thread.join()
producer_thread.join()