Python启航:30天编程速成之旅(第23天)- 多线程从入门到精通

liftword3周前 (02-07)技术文章19

喜欢的条友记得关注、点赞、转发、收藏，你们的支持就是我最大的动力源泉。

前期基础教程：

「Python3.11.0」手把手教你安装最新版Python运行环境

讲讲Python环境使用Pip命令快速下载各类库的方法

Python启航：30天编程速成之旅（第2天）-IDE安装

【Python教程】JupyterLab 开发环境安装

Python启航：30天编程速成之旅（第23天）- 多线程从入门到精通

简介

什么是多线程？

多线程是指一个程序中可以同时运行多个线程。每个线程是程序的一个独立执行路径，可以并行执行任务。多线程允许多个任务在同一个进程中并发执行，从而提高程序的效率和响应速度。

为什么使用多线程？

  1. 提高程序的响应性：例如，在 GUI 应用中，主线程负责处理用户界面，而其他线程可以执行后台任务，确保用户界面不会因为长时间的任务而卡住。
  2. 充分利用多核 CPU：虽然 Python 的 GIL 限制了多线程在 CPU 密集型任务中的并行性，但对于 I/O 密集型任务（如网络请求、文件读写），多线程可以显著提高性能。
  3. 简化代码结构：通过将复杂的任务分解为多个线程，可以使代码更易于理解和维护。

Python 中的 GIL（全局解释器锁）

Python 的 CPython 解释器有一个称为 GIL（Global Interpreter Lock） 的机制，它确保同一时刻只有一个线程在执行 Python 字节码。这意味着即使你有多个 CPU 核心，Python 的多线程也无法真正实现 CPU 密集型任务的并行执行。

然而，对于 I/O 密集型任务（如网络请求、文件读写），GIL 的影响较小，因为这些任务通常会释放 GIL 以等待 I/O 操作完成。因此，多线程在 I/O 密集型任务中仍然非常有用。

初级：创建和启动线程

使用threading.Thread创建线程

Python 提供了 threading 模块来处理多线程。最简单的方式是使用 threading.Thread 类来创建和启动线程。

import threading
import time

def print_numbers():
    for i in range(5):
        print(f"Number: {i}")
        time.sleep(1)

def print_letters():
    for letter in 'ABCDE':
        print(f"Letter: {letter}")
        time.sleep(1)

# 创建两个线程
t1 = threading.Thread(target=print_numbers)
t2 = threading.Thread(target=print_letters)

# 启动线程
t1.start()
t2.start()

# 等待线程结束
t1.join()
t2.join()

print("Both threads have finished.")

线程的基本属性和方法

  • start()：启动线程，调用线程的目标函数。
  • join()：阻塞当前线程，直到目标线程结束。这通常用于确保主线程等待所有子线程完成。
  • is_alive()：检查线程是否正在运行。
  • name：获取或设置线程的名称。
  • daemon：设置线程是否为守护线程。守护线程会在主线程结束时自动终止。

线程的生命周期

线程的生命周期包括以下几个阶段：

  1. 新建：线程对象被创建，但尚未启动。
  2. 就绪：线程已经启动，等待调度器分配 CPU 时间。
  3. 运行：线程正在执行。
  4. 阻塞：线程暂时停止执行，等待某个条件（如 I/O 操作完成）。
  5. 死亡：线程执行完毕或被终止。

中级：线程同步与通信

当多个线程同时访问共享资源时，可能会导致数据不一致或竞争条件（race condition）。为了防止这种情况，Python 提供了多种同步机制。

锁（Lock）与互斥锁（RLock）

Lock 是最简单的同步原语，用于确保一次只有一个线程可以访问共享资源。

import threading
import time

lock = threading.Lock()

def thread_function(name):
    with lock:
        print(f"Thread {name} is acquiring the lock.")
        time.sleep(1)
        print(f"Thread {name} is releasing the lock.")

# 创建多个线程
threads = []
for i in range(3):
    t = threading.Thread(target=thread_function, args=(i,))
    threads.append(t)
    t.start()

# 等待所有线程结束
for t in threads:
    t.join()

RLock（可重入锁）允许同一线程多次获取锁，而不会导致死锁。

rlock = threading.RLock()

with rlock:
    # 可以再次获取同一锁
    with rlock:
        print("This is safe with RLock.")

条件变量（Condition）

Condition 对象用于在线程之间进行更复杂的通信。它可以用来等待某个条件成立，或者通知其他线程条件已经满足。

import threading
import time

condition = threading.Condition()
data = []

def producer():
    for i in range(5):
        with condition:
            data.append(i)
            print(f"Produced: {i}")
            condition.notify()  # 通知消费者
        time.sleep(1)

def consumer():
    for _ in range(5):
        with condition:
            condition.wait()  # 等待生产者
            item = data.pop(0)
            print(f"Consumed: {item}")

# 创建生产者和消费者线程
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)

t1.start()
t2.start()

t1.join()
t2.join()

事件（Event）

Event 对象用于在线程之间传递简单的信号。一个线程可以设置或清除事件，另一个线程可以等待事件的发生。

import threading
import time

event = threading.Event()

def wait_for_event():
    print("Waiting for event...")
    event.wait()  # 等待事件发生
    print("Event occurred!")

def set_event():
    time.sleep(3)
    print("Setting event...")
    event.set()  # 触发事件

# 创建线程
t1 = threading.Thread(target=wait_for_event)
t2 = threading.Thread(target=set_event)

t1.start()
t2.start()

t1.join()
t2.join()

队列（Queue）

Queue 是一个线程安全的队列，适用于生产者-消费者模式。生产者线程将数据放入队列，消费者线程从队列中取出数据。

from queue import Queue
import threading
import time

queue = Queue()

def producer(queue):
    for i in range(5):
        queue.put(i)
        print(f"Produced: {i}")
        time.sleep(1)

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        queue.task_done()

# 创建生产者和消费者线程
t1 = threading.Thread(target=producer, args=(queue,))
t2 = threading.Thread(target=consumer, args=(queue,))

t1.start()
t2.start()

t1.join()
queue.put(None)  # 发送终止信号给消费者
t2.join()

高级：线程池与并发编程

使用concurrent.futures模块

concurrent.futures 模块提供了一个高层次的接口来管理线程池和进程池。它简化了并发编程，尤其是当你需要提交多个任务时。

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    print(f"Task {n} started")
    time.sleep(1)
    return f"Task {n} completed"

# 创建线程池
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交多个任务
    futures = [executor.submit(task, i) for i in range(5)]
    
    # 获取任务结果
    for future in futures:
        print(future.result())

线程池（ThreadPoolExecutor）

ThreadPoolExecutor 是 concurrent.futures 模块中的一个类，用于管理线程池。你可以指定最大线程数，并提交多个任务给线程池执行。

from concurrent.futures import ThreadPoolExecutor
import time

def download_file(url):
    print(f"Downloading {url}...")
    time.sleep(2)
    return f"{url} downloaded"

urls = [
    "https://example.com/file1",
    "https://example.com/file2",
    "https://example.com/file3"
]

# 创建线程池
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务
    results = list(executor.map(download_file, urls))
    
    # 打印结果
    for result in results:
        print(result)

异步 I/O 与asyncio

对于 I/O 密集型任务，asyncio 提供了更高效的异步编程模型。asyncio 基于协程（coroutine），可以在单线程中实现并发操作。

import asyncio

async def fetch_data(url):
    print(f"Fetching {url}...")
    await asyncio.sleep(2)  # 模拟网络请求
    return f"{url} fetched"

async def main():
    urls = [
        "https://example.com/file1",
        "https://example.com/file2",
        "https://example.com/file3"
    ]
    
    # 并发执行多个任务
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    # 打印结果
    for result in results:
        print(result)

# 运行异步主函数
asyncio.run(main())

并发模式：生产者-消费者模型

生产者-消费者模型是一种常见的并发模式，适用于多个生产者生成数据，多个消费者处理数据的场景。Queue 和 asyncio.Queue 都可以用于实现这种模式。

from queue import Queue
import threading
import time

queue = Queue()

def producer(queue):
    for i in range(5):
        queue.put(i)
        print(f"Produced: {i}")
        time.sleep(1)

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        queue.task_done()

# 创建生产者和消费者线程
t1 = threading.Thread(target=producer, args=(queue,))
t2 = threading.Thread(target=consumer, args=(queue,))

t1.start()
t2.start()

t1.join()
queue.put(None)  # 发送终止信号给消费者
t2.join()

性能优化与最佳实践

减少锁的竞争

锁的使用会导致线程之间的竞争，降低性能。尽量减少锁的使用范围，只在必要时加锁，并且尽量缩短持有锁的时间。

lock = threading.Lock()

def update_shared_resource(shared_resource, value):
    with lock:
        # 尽量减少锁的持有时间
        shared_resource += value

使用multiprocessing模块绕过 GIL

对于 CPU 密集型任务，multiprocessing 模块可以通过创建多个进程来绕过 GIL 的限制。每个进程都有自己的 Python 解释器和内存空间，因此可以真正实现并行执行。

from multiprocessing import Pool

def cpu_intensive_task(x):
    return sum(i * i for i in range(x))

if __name__ == '__main__':
    with Pool(processes=4) as pool:
        results = pool.map(cpu_intensive_task, [10000, 20000, 30000, 40000])
        print(results)

线程安全的数据结构

Python 提供了一些线程安全的数据结构，如 queue.Queue、threading.local 等。使用这些数据结构可以避免手动加锁，简化代码。

from queue import Queue

queue = Queue()

def producer(queue):
    for i in range(5):
        queue.put(i)
        print(f"Produced: {i}")
        time.sleep(1)

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        queue.task_done()

性能分析与调试

使用 cProfile 或 line_profiler 等工具可以帮助你分析代码的性能瓶颈。对于多线程程序，还可以使用 threading.settrace() 来跟踪线程的执行情况。

import cProfile
import pstats

def profile_code():
    # 你的代码
    pass

profiler = cProfile.Profile()
profiler.enable()
profile_code()
profiler.disable()

stats = pstats.Stats(profiler).sort_stats('cumulative')
stats.print_stats()

常见问题与解决方案

死锁（Deadlock）

死锁是指两个或多个线程互相等待对方释放资源，导致它们都无法继续执行。为了避免死锁，尽量避免嵌套锁的使用，或者使用 try...finally 语句确保锁总是会被释放。

lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    with lock1:
        time.sleep(1)
        with lock2:
            print("Thread 1 done")

def thread2():
    with lock2:
        time.sleep(1)
        with lock1:
            print("Thread 2 done")

# 这种情况下可能会发生死锁
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

t1.start()
t2.start()

t1.join()
t2.join()

活锁（Livelock）

活锁是指线程不断重复尝试执行某个操作，但由于条件始终不满足，导致它们无法继续前进。为了避免活锁，可以在每次尝试失败后引入随机延迟，或者使用超时机制。

import random

def livelock_example():
    while True:
        if not can_acquire_lock():
            time.sleep(random.random())  # 随机延迟
        else:
            break

线程饥饿（Thread Starvation）

线程饥饿是指某些线程由于优先级较低或其他原因，长期无法获得 CPU 时间。为了避免线程饥饿，可以使用公平锁（Fair Lock），或者确保高优先级线程不会长时间占用资源。

from threading import Lock

fair_lock = Lock()

def fair_thread():
    with fair_lock:
        print("Fair thread acquired the lock")

线程安全的第三方库一些第三方库提供了线程安全的功能，例如 requests.Session、pandas.DataFrame 等。在使用这些库时，确保了解它们的线程安全性，避免不必要的锁竞争。

实战案例

网络爬虫中的多线程应用

网络爬虫通常需要从多个网站抓取数据，这是一个典型的 I/O 密集型任务。使用多线程可以显著提高爬虫的效率。

import requests
from concurrent.futures import ThreadPoolExecutor

def fetch_url(url):
    response = requests.get(url)
    return response.text

喜欢的条友记得关注、点赞、转发、收藏，你们的支持就是我最大的动力源泉。

相关文章

Python零基础到精通，这8个入门技巧让你少走弯路，7天速通编程!

Python学习就像玩积木，从最基础的块开始，一步步搭建出复杂的作品。我记得刚开始学Python时也是一头雾水，走了不少弯路。现在回头看，其实掌握几个核心概念，就能快速入门这门编程语言。来聊聊怎么用最...

史上最详细python学习路线-从入门到精通，只需5个月时间

针对Python的初学者，从无到有的Python语言如何入门，主要包括了：Python的简介，如何下载Python，如何安装Python，如何使用终端、Shell，IDE等各种开发环境进行Python...

【2022·合辑】Python量化从入门到精通

引言春去秋来，公众号“Python金融量化”已坚持了四个年头，全网关注量10万+。在公众号阅读量急剧下降，短视频盛行的当下仍坚持文字输出着实不容易。这一路走来非常感谢广大读者的支持，尤其是感谢知识星球...

Python 自学秘籍:从入门到精通的奇妙之旅

在当今数字化的时代，Python 这门编程语言可谓是大放异彩，越来越多的小伙伴们都渴望掌握它，实现自己的各种梦想。那么，问题来了，怎么自学 Python 呢？自学这门神奇的语言又大概需要多久呢？今天，...

Python 从入门到精通知识点梳理_《python从入门到精通》

一、Python 基础1. Python 简介Python 特点：解释型、动态类型、面向对象、跨平台。应用领域：Web 开发、数据分析、人工智能、自动化运维、科学计算等。安装与环境配置：安装 Pyth...

神仙级python入门教程(非常详细)，从零基础到精通，从看这篇开始

一、初识Python的魅力Python的学习意义在学习Python之前，你不要担心自己没基础或“脑子笨”，我始终认为，只要你想学并为之努力，就能学好，就能用Python去做很多事情。在这个喧嚣的时代，...

蜀ICP备2024111239号-1