一、多线程前置基础概念

1.进程与线程区别

进程是操作系统资源分配最小单位，每个进程独立占用内存、CPU资源；线程依附于进程，是CPU调度最小单位，同一进程内所有线程共享进程内存、全局变量，创建销毁开销远小于进程。

• 多线程适用：IO密集型任务（网络爬虫、接口请求、文件读写、数据库查询），大量时间处于IO阻塞等待；
• 多线程不适用：CPU密集型（数值运算、矩阵计算），受Python GIL全局解释器锁限制，同一时刻仅有1个线程执行Python字节码，无法利用多核，这类场景改用multiprocessing多进程。

2.GIL全局解释器锁通俗说明

CPython自带GIL锁，是Python原生多线程无法多核并行的核心原因。IO阻塞时GIL会自动释放，因此IO任务多线程提速明显；纯计算场景多线程反而因为线程切换损耗性能，效率不如单线程。

3.threading模块介绍

threading是Python标准内置库，无需pip安装，原生实现线程创建、线程等待、守护线程、线程锁、线程通信全套能力，是Python多线程开发底层首选，比老旧_thread模块功能完善、稳定性更高。

二、两种创建线程的标准写法（项目最常用）

方式1：函数传参创建线程（简单任务首选）

把业务逻辑封装成普通函数，通过threading.Thread(target=函数, args=元组参数, kwargs=字典参数)创建线程对象，必须调用start()启动线程，直接调用函数是主线程串行执行。

import threading
import time

# 定义线程执行函数
def work(task_name, sleep_sec):
    print(f"【{task_name}】线程启动，当前线程名：{threading.current_thread().name}")
    time.sleep(sleep_sec)  # 模拟IO等待（爬虫、读文件）
    print(f"【{task_name}】线程执行完毕")

if __name__ == "__main__":
    start_time = time.time()
    # 创建2个子线程
    t1 = threading.Thread(target=work, args=("任务A", 2), name="Thread-A")
    t2 = threading.Thread(target=work, args=("任务B", 1), name="Thread-B")
    # 启动线程
    t1.start()
    t2.start()
    # join()：主线程阻塞，等待对应子线程执行结束再向下运行
    t1.join()
    t2.join()
    print(f"全部任务结束，总耗时：{round(time.time()-start_time,2)}s")

运行结果：总耗时约2秒，两个任务并发执行，而非串行3秒，直观体现多线程提速效果。

方式2：继承Thread类自定义线程（复杂业务封装首选）

继承threading.Thread，重写__init__构造方法与run()方法，run内部为线程核心执行逻辑，适合多步骤、需要封装属性的复杂任务。

import threading
import time

class CustomThread(threading.Thread):
    # 构造函数接收自定义参数
    def __init__(self, task_name, delay):
        super().__init__()  # 必须调用父类构造
        self.task_name = task_name
        self.delay = delay

    # 重写run：线程start后自动执行run方法
    def run(self):
        print(f"自定义线程：{self.task_name}开始运行")
        time.sleep(self.delay)
        print(f"自定义线程：{self.task_name}运行结束")

if __name__ == "__main__":
    t1 = CustomThread("自定义任务1", 2)
    t2 = CustomThread("自定义任务2", 1)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("所有自定义线程执行完成")

三、核心常用方法详解

1.start()

就绪线程，向操作系统申请CPU调度，只能调用一次，重复调用抛RuntimeError。

2.join(timeout=None)

• 无timeout：主线程永久阻塞，直到目标子线程执行完毕；
• 带数字timeout：主线程最多等待N秒，超时后不再阻塞，继续向下执行；

# 等待线程最多1.5秒，超时自动放行主线程
t1.join(timeout=1.5)

3.线程信息获取函数

import threading
# 获取当前运行线程对象
cur_thread = threading.current_thread()
print("当前线程名称：", cur_thread.name)
print("当前线程ID：", cur_thread.ident)
# 获取当前存活的线程总数（含主线程）
print("活跃线程数量：", threading.active_count())
# 查看所有存活线程列表
print(threading.enumerate())

四、守护线程Daemon（后台常驻任务必备）

概念

非守护线程（默认daemon=False）：主线程执行完代码后，会等待所有子线程全部结束，程序才退出；
守护线程（daemon=True）：主线程代码执行完毕，无论守护线程是否运行完毕，强制跟着主线程退出，多用于日志监听、定时巡检、心跳检测等后台辅助任务。

注意：daemon属性必须在start()调用前设置，启动后修改直接报错。

import threading
import time

def daemon_task():
    while True:
        print("守护线程正在后台运行...")
        time.sleep(1)

def normal_task():
    time.sleep(3)
    print("普通子线程执行结束")

if __name__ == "__main__":
    t1 = threading.Thread(target=daemon_task, daemon=True)
    t2 = threading.Thread(target=normal_task)
    t1.start()
    t2.start()
    time.sleep(2)
    print("主线程代码执行完毕，程序退出")

运行说明：主线程休眠2秒结束，守护线程立刻终止；普通子线程还需1秒才结束，程序会继续等待普通线程跑完才退出。

五、线程同步锁：解决共享资源数据错乱

多线程共用全局变量时，count +=1这类非原子操作会出现数据竞争（竞态条件），多个线程同时读取、修改变量，最终数值小于预期，需要锁保证同一时间只有一个线程操作临界资源。

1.Lock互斥锁（最常用）

两种写法：手动acquire/release、with自动上下文（推荐，异常自动释放锁，避免死锁）

import threading

count = 0
lock = threading.Lock()  # 创建锁对象

def add_num():
    global count
    for _ in range(100000):
        # 写法1：with自动获取释放锁
        with lock:
            count +=1
        
        # 写法2：手动上锁解锁（不推荐，异常容易丢锁死锁）
        # lock.acquire()
        # try:
        #     count +=1
        # finally:
        #     lock.release()

if __name__ == "__main__":
    thread_list = []
    for i in range(5):
        t = threading.Thread(target=add_num)
        thread_list.append(t)
        t.start()
    for t in thread_list:
        t.join()
    print("最终统计数值：", count) # 加锁后固定500000，不加锁数值随机偏小

2.RLock可重入锁

同一个线程可以多次acquire上锁，需要同等次数release解锁，适用于嵌套加锁场景（Lock不可重入，同一线程重复上锁直接死锁）。

r_lock = threading.RLock()
def test_rlock():
    r_lock.acquire()
    print("第一层上锁")
    r_lock.acquire()
    print("第二层嵌套上锁")
    r_lock.release()
    r_lock.release()

六、线程间安全通信：queue队列

多线程不建议直接用全局变量传数据，队列queue.Queue是线程安全容器，自带锁机制，生产消费模型标准方案：生产者线程put数据、消费者线程get数据。

import threading
import queue
import time

# 创建先进先出队列
task_queue = queue.Queue(maxsize=10)

# 生产者：往队列写入数据
def producer():
    for i in range(8):
        task_queue.put(f"数据-{i}")
        print(f"生产：数据-{i}")
        time.sleep(0.2)

# 消费者：从队列取出处理
def consumer():
    while True:
        data = task_queue.get() # 队列空时阻塞等待
        print(f"消费：{data}")
        task_queue.task_done() # 标记单个任务处理完成
        time.sleep(0.5)

if __name__ == "__main__":
    t_pro = threading.Thread(target=producer)
    t_con = threading.Thread(target=consumer, daemon=True)
    t_pro.start()
    t_con.start()
    t_pro.join()
    task_queue.join() # 等待队列所有数据消费完毕
    print("所有数据生产消费完成")

七、实战案例：多线程批量爬虫（项目落地）

使用threading+Queue实现批量网页请求，模拟爬虫并发抓取页面，是日常开发高频场景。

import threading
import queue
import requests

headers = {"User-Agent":"Mozilla/5.0"}
url_queue = queue.Queue()
# 待爬取链接列表
url_list = [
    "https://www.baidu.com",
    "https://www.bing.com",
    "https://www.qq.com"
]
# 链接入队
for url in url_list:
    url_queue.put(url)

def crawl():
    while not url_queue.empty():
        target_url = url_queue.get()
        try:
            resp = requests.get(target_url,headers=headers,timeout=5)
            print(f"【{threading.current_thread().name}】{target_url} 状态码：{resp.status_code}，页面长度：{len(resp.text)}")
        except Exception as e:
            print(f"爬取失败{target_url}，错误：{e}")
        url_queue.task_done()

if __name__ == "__main__":
    thread_num = 3 # 开启3个爬虫线程
    thread_list = []
    for i in range(thread_num):
        t = threading.Thread(target=crawl,name=f"爬虫线程{i+1}")
        thread_list.append(t)
        t.start()
    # 等待队列全部处理完
    url_queue.join()
    print("全部链接爬取结束")