python中的线程和线程池

线程示例

importthreadingimporttimedefworker(msg):print(f"Working on{msg}")time.sleep(1)print(f"Finished{msg}")# 1. 创建线程对象t1=threading.Thread(target=worker,args=("Task-1",))t2=threading.Thread(target=worker,args=("Task-2",))# 2. 启动线程t1.start()t2.start()# 3. 【重要】等待线程执行完毕 (join)# 这样主程序会阻塞在这里，直到 t1 和 t2 跑完，避免主程序提前退出导致子线程被截断t1.join()t2.join()print("所有任务完成！")

线程池示例(简单)

fromconcurrent.futuresimportThreadPoolExecutorimporttimeimportrequestsdeffetch_data(url:str)->str:"""模拟IO密集型任务"""print(f"正在请求:{url}")time.sleep(1)# 模拟网络延迟returnf"来自{url}的数据"if__name__=="__main__":urls=[f"https://example.com/{i}"foriinrange(5)]# max_workers:# Python 3.13 默认值通常是 min(32, os.cpu_count() + 4)# 对于IO密集型，建议手动设置大一点，比如 20-50withThreadPoolExecutor(max_workers=10)asexecutor:# map 会按输入顺序返回结果results=executor.map(fetch_data,urls)forurl,resultinzip(urls,results):print(f"{url}:{result}")print("所有任务完成！")

线程池示例(需要响应的)

fromconcurrent.futuresimportThreadPoolExecutor,as_completedimporttimeimportrandomdefprocess_task(task_id:int)->str:duration=random.uniform(0.5,2.0)time.sleep(duration)iftask_id==3:raiseValueError(f"任务{task_id}故意失败")returnf"任务{task_id}完成 (耗时{duration:.2f}s)"if__name__=="__main__":withThreadPoolExecutor(max_workers=5)asexecutor:# 提交任务并记录 Future 对象future_to_id={executor.submit(process_task,i):iforiinrange(6)}# as_completed 会在任务完成时立即返回，不等待其他慢任务forfutureinas_completed(future_to_id):task_id=future_to_id[future]try:result=future.result()print(f"✅ 成功:{result}")exceptExceptionase:print(f"❌ 任务{task_id}失败:{e}")print("主程序结束")