python创建线程和结束线程
'# Python创建线程和结束线程
一、背景与问题
在Python中实现多线程是并发编程的常见需求，但其底层机制与其它语言存在本质差异。Python的全局解释器锁（GIL）机制决定了线程的执行方式，这使得多线程在CPU密集型任务中表现有限，但适合IO密集型场景。
多线程编程需要处理以下核心问题：
- 线程创建与生命周期管理
- 线程间通信与同步
- 异常处理与资源释放
- 线程终止的优雅方式
- 性能瓶颈与优化手段
二、基本原理
1. Python线程机制
Python的threading模块基于底层的_thread库实现。线程创建时会：
- 申请一个新的线程标识符
- 分配独立的栈空间
- 将线程加入线程调度队列
- 通过GIL控制CPU时间片分配
GIL（Global Interpreter Lock）是Python的互斥锁，确保同一时刻只有一个线程执行Python字节码。这意味着多线程在CPU密集型任务中无法实现真正的并行，但IO操作时可利用多线程进行并发。
2. 线程生命周期
线程生命周期包含：
- 创建（
Thread实例化） - 启动（
start()方法） - 执行（
run()方法） - 等待（
join()方法） - 终止（正常退出或异常终止）
3. 线程间通信
线程间通信主要通过以下机制实现：
Event对象：事件通知Condition对象：条件变量Queue队列：线程安全队列Lock锁：互斥锁Semaphore信号量：资源控制
三、环境准备
# 安装必要的依赖包（如需）
pip install requests核心模块导入：
import threading
import time
import requests四、核心实现
示例1：基本线程创建与启动
import threading
import time
def worker(name):
print(f"Thread {name} started")
time.sleep(2)
print(f"Thread {name} finished")
# 创建线程
t1 = threading.Thread(target=worker, args=("Thread-1",))
t2 = threading.Thread(target=worker, args=("Thread-2",))
# 启动线程
t1.start()
t2.start()
# 等待线程完成
t1.join()
t2.join()
print("All threads completed")关键代码解释：
Thread构造函数接受target（目标函数）和args（参数）start()方法会自动调用run()方法join()会阻塞主线程直到指定线程完成- 线程默认是非守护线程（
daemon=False），主线程会等待其完成
示例2：使用Thread类的run方法
import threading
import time
class MyThread(threading.Thread):
def __init__(self, name):
super().__init__()
self.name = name
def run(self):
print(f"Thread {self.name} started")
time.sleep(2)
print(f"Thread {self.name} finished")
# 创建并启动线程
t1 = MyThread("Thread-1")
t2 = MyThread("Thread-2")
t1.start()
t2.start()
t1.join()
t2.join()关键代码解释：
- 重写
run()方法定义线程执行逻辑 super().__init__()确保继承正确初始化- 线程启动后会自动调用
run()方法 join()确保主线程等待子线程完成
示例3：守护线程与超时处理
import threading
import time
import requests
def fetch_url(url, timeout=10):
try:
response = requests.get(url, timeout=timeout)
print(f"Fetch {url} success, status code: {response.status_code}")
except Exception as e:
print(f"Fetch {url} error: {str(e)}")
# 创建守护线程
daemon_thread = threading.Thread(
target=fetch_url,
args=("https://httpbin.org/get",),
daemon=True
)
# 启动线程
daemon_thread.start()
# 设置超时并等待
daemon_thread.join(timeout=5)
print("Main thread done")关键代码解释：
daemon=True设置守护线程，主线程退出后自动终止join(timeout=5)设置等待超时，避免无限等待- 网络请求超时由requests库处理，但需要捕获异常
五、完整案例
多文件下载器案例
import threading
import requests
import time
from concurrent.futures import ThreadPoolExecutor
def download_file(url, filename):
try:
response = requests.get(url, timeout=10)
with open(filename, 'wb') as f:
f.write(response.content)
print(f"Downloaded {filename} successfully")
except Exception as e:
print(f"Download {filename} error: {str(e)}")
def main():
urls = [
"https://httpbin.org/get",
"https://httpbin.org/post",
"https://httpbin.org/bytes/1024"
]
# 使用线程池控制并发数
with ThreadPoolExecutor(max_workers=3) as executor:
# 提交任务
futures = []
for i, url in enumerate(urls):
filename = f"file_{i}.txt"
future = executor.submit(download_file, url, filename)
futures.append(future)
# 等待所有任务完成
for future in futures:
future.result()
if __name__ == "__main__":
start_time = time.time()
main()
print(f"Total time: {time.time() - start_time:.2f}s")关键代码解释：
- 使用
ThreadPoolExecutor控制最大并发数 submit()方法提交任务并返回Future对象result()方法获取任务结果并处理异常- 线程池自动管理线程生命周期
六、源码解析
threading模块源码关键点
线程类定义：
class Thread(_Thread): def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, daemon=None): # 初始化线程对象 if daemon is not None: self.daemon = daemon # 其他初始化代码线程启动机制：
def start(self): # 检查线程是否已启动 if self._is_stopped: raise RuntimeError("Thread already started") # 创建线程并启动 self._Thread__started = True self._Thread__stop = False self._Thread__lock = _allocate_lock() self._Thread__ident = _get_ident() # 将线程加入调度队列 _start_new_thread(self._Thread__bootstrap, ())线程终止机制：
def join(self, timeout=None): # 等待线程完成 if self._Thread__stopped: return # 处理超时逻辑 if timeout is not None: deadline = time.time() + timeout while True: if self._Thread__stopped: return # 等待线程完成 time.sleep(0.1) if timeout is not None and time.time() > deadline: raise TimeoutError("Thread timed out")
七、进阶使用
1. 线程池优化
from concurrent.futures import ThreadPoolExecutor
def worker(n):
print(f"Processing {n}")
time.sleep(1)
return n * 2
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(worker, range(10)))
print(results)2. 线程间通信
import threading
import time
event = threading.Event()
def worker():
print("Worker waiting for event")
event.wait()
print("Worker received event")
t = threading.Thread(target=worker)
t.start()
time.sleep(1)
event.set()3. 线程安全队列
from queue import Queue
q = Queue()
def worker():
while True:
item = q.get()
if item is None:
break
print(f"Processing {item}")
q.task_done()
t = threading.Thread(target=worker)
t.start()
for i in range(5):
q.put(i)
q.join()八、性能与工程实践
1. 性能瓶颈分析
| 场景 | 建议方案 | 原因 |
|---|---|---|
| CPU密集型 | 多进程 | GIL限制多线程并发 |
| IO密集型 | 线程 | 并发IO操作 |
| 网络请求 | 异步IO | 避免阻塞主线程 |
2. 线程终止优化
- 使用
threading.Event控制线程退出 - 设置超时机制避免死锁
- 使用
join(timeout)控制等待时间
3. 线程安全注意事项
- 竞态条件：使用锁（
Lock/RLock）保护共享资源 - 数据竞争：使用
Queue替代直接共享变量 - 死锁：遵循锁获取顺序，使用
with语句
4. 异常处理
def safe_worker():
try:
# 可能引发异常的代码
except Exception as e:
# 异常处理逻辑
print(f"Caught exception: {str(e)}")九、常见问题与踩坑
常见问题列表
- 主线程未等待子线程：未使用
join()导致资源未释放 - 守护线程未及时终止：未设置
daemon=True导致主线程等待 - 死锁问题：多个锁的获取顺序不一致
- 资源泄漏：未正确关闭文件/网络连接
- GIL限制：CPU密集型任务效率低下
常见错误示例
# 错误示例：未处理异常导致线程终止
def bad_worker():
print("Starting worker")
time.sleep(5)
print("Worker done")
t = threading.Thread(target=bad_worker)
t.start()改进方法：
- 添加异常处理
- 使用
Thread.join(timeout)控制超时 - 使用
ThreadPoolExecutor管理线程池
十、最佳实践
推荐方案选择
| 场景 | 推荐方案 | 说明 |
|---|---|---|
| IO密集型任务 | 线程 | 并发IO操作 |
| CPU密集型任务 | 多进程 | 避免GIL限制 |
| 异步IO | asyncio | 非阻塞IO操作 |
| 资源密集型 | 线程池 | 控制并发数量 |
| 跨平台 | concurrent.futures | 统一接口 |
资源管理建议
- 使用
with语句管理文件/网络资源 - 使用
contextlib管理上下文 - 使用
atexit注册清理函数
线程终止建议
- 使用
Event或Condition控制线程退出 - 设置超时机制避免死锁
- 使用
ThreadPoolExecutor自动管理线程生命周期
十一、总结
Python线程编程需要深入理解GIL机制和线程调度原理。虽然多线程在CPU密集型任务中表现有限，但在IO密集型场景下可以显著提升并发性能。实际开发中应根据任务类型选择合适方案：IO密集型使用线程，CPU密集型使用多进程，异步IO使用asyncio。需要特别注意线程安全、异常处理、资源管理和性能优化，避免常见的死锁、资源泄漏和GIL限制等问题。通过合理使用线程池、守护线程和同步机制，可以构建高效稳定的并发系统。
评论已关闭