Python进阶笔记全攻略：从面向对象到并发编程

        作为一门兼具简洁性和强大性的编程语言，Python 的进阶特性是从 “会用” 到 “用好” 的核心门槛。本文将系统梳理 Python 进阶的核心知识点，涵盖面向对象编程、高级语法、数据库操作、设计模式、并发编程等模块，每个知识点均配套可运行的示例代码

一、面向对象编程（OOP）：Python 的核心编程范式

面向对象是 Python 一切高级特性的基础，其核心是 “万物皆对象”，通过类（Class）封装属性和行为，通过对象（Instance）实现具体功能。

1. 类与对象：OOP 的基本单元

类：是具有相同属性和行为的对象的抽象模板，定义了对象的 “蓝图”。

对象：是类的具体实例，是内存中实际存在的实体。

构造方法：__init__ 是 Python 的构造方法，用于初始化对象的属性，创建对象时自动调用。

示例代码：定义类并创建对象。

# 定义一个“人”的类
class Person:
# 构造方法：初始化属性
def __init__(self, name: str, age: int):
# 实例属性：每个对象独有的属性
self.name = name
self.age = age

# 实例方法：描述对象的行为
def introduce(self) -> None:
"""自我介绍"""
print(f"大家好，我是{self.name}，今年{self.age}岁。")

# 创建类的实例（对象）
person1 = Person("张三", 25)
person2 = Person("李四", 30)

# 调用对象的方法
person1.introduce() # 输出：大家好，我是张三，今年25岁。
person2.introduce() # 输出：大家好，我是李四，今年30岁。

# 访问对象的属性
print(person1.name) # 输出：张三
print(person2.age) # 输出：30

2. 方法：类的行为定义

Python 中的方法分为三类：实例方法、类方法、静态方法，三者的核心区别是第一个参数的类型和绑定关系。

实例方法：self（代表当前对象），绑定到对象，对象.方法 ()。

类方法：cls（代表当前类），绑定到类，类.方法 () / 对象.方法 ()。

静态方法：无默认参数，无绑定关系，类.方法 () / 对象.方法 ()。

示例代码：三种方法的使用

class Student:
# 类属性：所有对象共享的属性
school = "北京大学"

def __init__(self, name: str, score: float):
self.name = name
self.score = score

# 1. 实例方法：操作实例属性
def get_score(self) -> float:
"""获取学生成绩"""
return self.score

# 2. 类方法：用@classmethod装饰，操作类属性
@classmethod
def change_school(cls, new_school: str) -> None:
"""修改学校名称（类属性）"""
cls.school = new_school

# 3. 静态方法：用@staticmethod装饰，无绑定关系，类似普通函数
@staticmethod
def is_excellent(score: float) -> bool:
"""判断是否为优秀成绩（与类/对象属性无关）"""
return score >= 90

# 实例方法调用
stu = Student("王五", 95)
print(stu.get_score()) # 输出：95

# 类方法调用（推荐用类调用）
Student.change_school("清华大学")
print(Student.school) # 输出：清华大学
print(stu.school) # 输出：清华大学（类属性共享）

# 静态方法调用（推荐用类调用）
print(Student.is_excellent(85)) # 输出：False
print(Student.is_excellent(92)) # 输出：True

3. 封装：隐藏内部实现，保障数据安全

封装是 OOP 的三大特性之一，核心思想是 “隐藏对象的内部细节，只暴露必要的接口”。Python 通过访问控制实现封装：

公开属性 / 方法：默认无下划线，可外部直接访问（如self.name）。

受保护属性 / 方法：单下划线开头（如_age），约定外部不直接访问（仅提醒，无强制限制）。

私有属性 / 方法：双下划线开头（如__salary），Python 会自动改名（名称改写），外部无法直接访问。

示例代码：封装的实现

class Employee:
def __init__(self, name: str, salary: float):
self.name = name # 公开属性
self._department = "技术部" # 受保护属性
self.__salary = salary # 私有属性

# 提供公开接口访问私有属性（getter）
def get_salary(self) -> float:
return self.__salary

# 提供公开接口修改私有属性（setter），可添加校验逻辑
def set_salary(self, new_salary: float) -> None:
if new_salary > 0:
self.__salary = new_salary
else:
raise ValueError("工资不能为负数")

# 私有方法：仅内部调用
def __calculate_bonus(self) -> float:
return self.__salary * 0.1

# 公开方法调用私有方法
def get_bonus(self) -> float:
return self.__calculate_bonus()

# 创建对象
emp = Employee("赵六", 10000)

# 访问公开属性
print(emp.name) # 输出：赵六

# 访问受保护属性（不推荐，但语法允许）
print(emp._department) # 输出：技术部

# 访问私有属性（直接访问会报错）
# print(emp.__salary) # AttributeError: 'Employee' object has no attribute '__salary'

# 通过公开接口访问私有属性
print(emp.get_salary()) # 输出：10000

# 通过公开接口修改私有属性（带校验）
emp.set_salary(12000)
print(emp.get_salary()) # 输出：12000

# 调用公开方法（内部调用私有方法）
print(emp.get_bonus()) # 输出：1200.0

# 尝试修改工资为负数（触发异常）
# emp.set_salary(-5000) # ValueError: 工资不能为负数

4. 继承：代码复用的核心手段

继承是 OOP 的三大特性之一，允许子类（Subclass）继承父类（Superclass）的属性和方法，子类可重写父类方法或扩展新功能，实现 “代码复用”。

语法：class 子类(父类1, 父类2,…):（Python 支持多继承）。

super()：调用父类的方法，解决多继承中的 MRO（方法解析顺序）问题。

示例代码：单继承与方法重写

# 父类：动物
class Animal:
def __init__(self, name: str):
self.name = name

def eat(self) -> None:
print(f"{self.name}正在吃东西")

def sleep(self) -> None:
print(f"{self.name}正在睡觉")

# 子类：狗（继承Animal）
class Dog(Animal):
# 重写父类的eat方法
def eat(self) -> None:
print(f"{self.name}正在啃骨头")

# 扩展新方法
def bark(self) -> None:
print(f"{self.name}正在汪汪叫")

# 子类：猫（继承Animal）
class Cat(Animal):
# 重写父类的__init__方法，调用父类构造方法
def __init__(self, name: str, color: str):
# 调用父类的__init__方法
super().__init__(name)
self.color = color # 扩展新属性

# 重写父类的eat方法
def eat(self) -> None:
print(f"{self.color}的{self.name}正在吃鱼")

# 创建子类对象
dog = Dog("旺财")
cat = Cat("咪咪", "白色")

# 调用继承的方法
dog.sleep() # 输出：旺财正在睡觉
cat.sleep() # 输出：咪咪正在睡觉

# 调用重写的方法
dog.eat() # 输出：旺财正在啃骨头
cat.eat() # 输出：白色的咪咪正在吃鱼

# 调用子类扩展的方法
dog.bark() # 输出：旺财正在汪汪叫

# 访问子类扩展的属性
print(cat.color) # 输出：白色

示例代码：多继承（Python 特有）

# 父类1：会飞
class Flyable:
def fly(self) -> None:
print("正在飞行")

# 父类2：会游泳
class Swimmable:
def swim(self) -> None:
print("正在游泳")

# 子类：鸭子（继承Flyable和Swimmable）
class Duck(Flyable, Swimmable):
def quack(self) -> None:
print("嘎嘎嘎")

# 创建鸭子对象
duck = Duck()

# 调用多个父类的方法
duck.fly() # 输出：正在飞行
duck.swim() # 输出：正在游泳
duck.quack() # 输出：嘎嘎嘎

# 查看MRO（方法解析顺序）：解决多继承的冲突问题
print(Duck.__mro__)
# 输出：(<class '__main__.Duck'>, <class '__main__.Flyable'>, <class '__main__.Swimmable'>, <class 'object'>)

5. 多态：灵活的行为扩展

多态是 OOP 的三大特性之一，核心思想是 “同一方法，不同对象有不同的行为”，无需关心对象的具体类型，只需调用统一的接口。Python 是动态类型语言，天然支持多态（无需像 Java 一样声明接口）。

示例代码：多态的实现

# 定义统一的接口函数
def make_sound(animal) -> None:
"""接收任意实现了sound方法的对象"""
animal.sound()

# 定义不同的类，都实现sound方法
class Dog:
def sound(self) -> None:
print("汪汪汪")

class Cat:
def sound(self) -> None:
print("喵喵喵")

class Bird:
def sound(self) -> None:
print("叽叽叽")

# 调用统一接口，传入不同对象
make_sound(Dog()) # 输出：汪汪汪
make_sound(Cat()) # 输出：喵喵喵
make_sound(Bird()) # 输出：叽叽叽

二、高级语法特性：提升代码简洁性与可读性

1. 类型注解：增强代码可读性与可维护性

类型注解（Type Hints）是 Python 3.5 + 引入的特性，用于标注变量、函数参数和返回值的类型，不影响代码运行，但可帮助开发者理解代码、辅助 IDE 提示、通过类型检查工具（如 mypy）发现错误。

示例代码：类型注解的全面使用

from typing import List, Dict, Tuple, Optional, Union

# 1. 变量注解
name: str = "张三"
age: int = 25
scores: List[float] = [90.5, 88.0, 95.0]
info: Dict[str, Union[str, int]] = {"name": "李四", "age": 30}
point: Tuple[int, int] = (10, 20)
optional_value: Optional[str] = None # 可选类型（str或None）

# 2. 函数注解
def calculate_sum(a: int, b: float) -> float:
"""计算两个数的和"""
return a + b

# 3. 类方法注解
class Calculator:
def multiply(self, x: List[int], y: int) -> List[int]:
"""将列表中的每个元素乘以y"""
return [num * y for num in x]

# 4. 返回值为多个类型的注解
def get_data() -> Union[int, str]:
"""返回整数或字符串"""
import random
return random.choice([100, "hello"])

# 调用函数（类型注解不影响运行）
print(calculate_sum(5, 3.2)) # 输出：8.2
calc = Calculator()
print(calc.multiply([1,2,3], 4)) # 输出：[4,8,12]
print(get_data()) # 随机输出100或hello

2. 装饰器：增强函数 / 方法的功能

装饰器（Decorator）是 Python 的高阶函数，用于在不修改原函数代码的前提下，为函数添加额外功能（如日志、计时、权限校验），是 “开闭原则” 的典型应用。

示例代码：自定义装饰器

import time
from functools import wraps

# 1. 基础装饰器：记录函数执行时间
def timer(func):
"""装饰器：记录函数执行时间"""
# wraps保留原函数的元信息（如__name__）
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"函数{func.__name__}执行耗时：{end_time – start_time:.4f}秒")
return result
return wrapper

# 2. 带参数的装饰器：日志装饰器
def logger(level: str = "INFO"):
"""带参数的装饰器：记录日志级别"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"[{level}] 函数{func.__name__}开始执行")
result = func(*args, **kwargs)
print(f"[{level}] 函数{func.__name__}执行完成")
return result
return wrapper
return decorator

# 使用装饰器
@timer
@logger(level="DEBUG") # 装饰器叠加（从下到上执行）
def slow_function(n: int):
"""模拟耗时函数"""
time.sleep(n)
return f"执行完成，休眠了{n}秒"

# 调用函数
print(slow_function(2))
# 输出：
# [DEBUG] 函数slow_function开始执行
# 函数slow_function执行耗时：2.0020秒
# [DEBUG] 函数slow_function执行完成
# 执行完成，休眠了2秒

3. 上下文管理器：优雅管理资源

上下文管理器（Context Manager）用于自动管理资源（如文件、数据库连接、网络连接），通过with语句实现 “进入时获取资源，退出时释放资源”，避免手动关闭资源导致的泄漏。

示例代码：自定义上下文管理器

# 方式1：通过类实现（__enter__和__exit__方法）
class FileManager:
def __init__(self, file_path: str, mode: str = "r"):
self.file_path = file_path
self.mode = mode
self.file = None

def __enter__(self):
"""进入上下文：打开文件"""
self.file = open(self.file_path, self.mode, encoding="utf-8")
return self.file

def __exit__(self, exc_type, exc_val, exc_tb):
"""退出上下文：关闭文件"""
if self.file:
self.file.close()
# 返回True表示忽略异常，False表示抛出异常
return False

# 方式2：通过contextlib.contextmanager装饰器实现（更简洁）
from contextlib import contextmanager

@contextmanager
def file_manager(file_path: str, mode: str = "r"):
"""上下文管理器：管理文件资源"""
try:
file = open(file_path, mode, encoding="utf-8")
yield file # 返回文件对象
finally:
file.close()

# 使用上下文管理器
# 方式1的使用
with FileManager("test.txt", "w") as f:
f.write("Hello, Python进阶！")

# 方式2的使用
with file_manager("test.txt", "r") as f:
content = f.read()
print(content) # 输出：Hello, Python进阶！

三、数据库操作：Python 与数据持久化

Python 支持多种数据库（SQLite、MySQL、PostgreSQL），核心是通过数据库驱动实现 “连接 – 操作 – 关闭”，推荐使用with语句管理连接，避免资源泄漏。

1. SQLite：轻量级内置数据库

SQLite 是 Python 内置的数据库，无需额外安装，适合小型应用、原型开发。

示例代码：SQLite 操作

import sqlite3
from contextlib import contextmanager

# 定义上下文管理器管理数据库连接
@contextmanager
def sqlite_connection(db_name: str):
"""SQLite连接管理器"""
conn = None
try:
# 连接数据库（不存在则创建）
conn = sqlite3.connect(db_name)
# 设置行工厂：返回字典格式的结果
conn.row_factory = sqlite3.Row
yield conn
conn.commit() # 提交事务
except Exception as e:
if conn:
conn.rollback() # 回滚事务
raise e
finally:
if conn:
conn.close() # 关闭连接

# 1. 创建表
with sqlite_connection("test.db") as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
age INTEGER NOT NULL,
email TEXT UNIQUE NOT NULL
)
""")

# 2. 插入数据
with sqlite_connection("test.db") as conn:
cursor = conn.cursor()
# 单条插入
cursor.execute(
"INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
("张三", 25, "zhangsan@example.com")
)
# 批量插入
users = [("李四", 30, "lisi@example.com"), ("王五", 28, "wangwu@example.com")]
cursor.executemany(
"INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
users
)

# 3. 查询数据
with sqlite_connection("test.db") as conn:
cursor = conn.cursor()
# 查询所有用户
cursor.execute("SELECT * FROM users")
for row in cursor.fetchall():
# 按列名访问数据
print(f"ID: {row['id']}, 姓名: {row['name']}, 年龄: {row['age']}, 邮箱: {row['email']}")

# 4. 更新数据
with sqlite_connection("test.db") as conn:
cursor = conn.cursor()
cursor.execute(
"UPDATE users SET age = ? WHERE name = ?",
(26, "张三")
)

# 5. 删除数据
with sqlite_connection("test.db") as conn:
cursor = conn.cursor()
cursor.execute("DELETE FROM users WHERE name = ?", ("王五",))

2. MySQL：主流关系型数据库

MySQL 是工业级关系型数据库，需安装pymysql驱动（pip install pymysql）。

示例代码：MySQL 操作

import pymysql
from contextlib import contextmanager
from typing import List, Dict

# 数据库配置
DB_CONFIG = {
"host": "localhost",
"port": 3306,
"user": "root",
"password": "your_password",
"database": "test_db",
"charset": "utf8mb4"
}

# 上下文管理器管理MySQL连接
@contextmanager
def mysql_connection():
"""MySQL连接管理器"""
conn = None
try:
conn = pymysql.connect(**DB_CONFIG)
yield conn
conn.commit()
except Exception as e:
if conn:
conn.rollback()
raise e
finally:
if conn:
conn.close()

# 1. 创建表
with mysql_connection() as conn:
with conn.cursor() as cursor:
cursor.execute("""
CREATE TABLE IF NOT EXISTS products (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
price DECIMAL(10,2) NOT NULL,
stock INT DEFAULT 0
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")

# 2. 插入数据
def insert_product(name: str, price: float, stock: int) -> None:
"""插入单个商品"""
with mysql_connection() as conn:
with conn.cursor() as cursor:
sql = "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)"
cursor.execute(sql, (name, price, stock))

# 调用插入函数
insert_product("Python进阶教程", 99.0, 1000)
insert_product("MySQL实战", 89.0, 500)

# 3. 查询数据
def get_products(min_price: float) -> List[Dict]:
"""查询价格大于等于min_price的商品"""
with mysql_connection() as conn:
with conn.cursor(pymysql.cursors.DictCursor) as cursor:
sql = "SELECT * FROM products WHERE price >= %s"
cursor.execute(sql, (min_price,))
return cursor.fetchall()

# 调用查询函数
products = get_products(90.0)
for product in products:
print(f"ID: {product['id']}, 名称: {product['name']}, 价格: {product['price']}, 库存: {product['stock']}")

# 4. 批量更新
def batch_update_stock(product_ids: List[int], new_stock: int) -> None:
"""批量更新商品库存"""
with mysql_connection() as conn:
with conn.cursor() as cursor:
sql = "UPDATE products SET stock = %s WHERE id IN (%s)"
# 构造IN条件的占位符
placeholders = ", ".join(["%s"] * len(product_ids))
sql = sql % ("%s", placeholders)
# 拼接参数：new_stock + product_ids
params = [new_stock] + product_ids
cursor.execute(sql, params)

# 调用批量更新函数
batch_update_stock([1, 2], 800)

四、设计模式：编写优雅可扩展的代码

设计模式是解决特定场景问题的成熟方案，Python 中常用的设计模式包括单例模式、工厂模式、装饰器模式等。

1. 单例模式：确保类只有一个实例

单例模式用于确保某个类在程序运行期间只有一个实例，适合管理全局资源（如配置、数据库连接池）。

# 方式1：基于__new__方法实现（最常用）
class Singleton:
_instance = None # 类属性存储唯一实例

def __new__(cls, *args, **kwargs):
if not cls._instance:
cls._instance = super().__new__(cls)
return cls._instance

def __init__(self, config: dict):
# 防止重复初始化
if not hasattr(self, "config"):
self.config = config

# 方式2：基于装饰器实现
def singleton_decorator(cls):
"""单例装饰器"""
instances = {}
def wrapper(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return wrapper

@singleton_decorator
class ConfigManager:
def __init__(self, config: dict):
self.config = config

# 测试方式1
s1 = Singleton({"db": "sqlite"})
s2 = Singleton({"db": "mysql"})
print(s1 is s2) # 输出：True
print(s1.config) # 输出：{'db': 'sqlite'}（初始化只执行一次）

# 测试方式2
cm1 = ConfigManager({"host": "localhost"})
cm2 = ConfigManager({"host": "127.0.0.1"})
print(cm1 is cm2) # 输出：True
print(cm1.config) # 输出：{'host': 'localhost'}

2. 工厂模式：封装对象的创建逻辑

工厂模式用于封装对象的创建过程，根据不同的条件返回不同的对象，降低代码耦合度。

# 定义产品接口（基类）
class Payment:
def pay(self, amount: float) -> str:
raise NotImplementedError("子类必须实现pay方法")

# 具体产品类：支付宝支付
class Alipay(Payment):
def pay(self, amount: float) -> str:
return f"使用支付宝支付{amount}元"

# 具体产品类：微信支付
class WeChatPay(Payment):
def pay(self, amount: float) -> str:
return f"使用微信支付{amount}元"

# 具体产品类：银行卡支付
class BankPay(Payment):
def pay(self, amount: float) -> str:
return f"使用银行卡支付{amount}元"

# 工厂类：创建支付对象
class PaymentFactory:
@staticmethod
def create_payment(pay_type: str) -> Payment:
"""根据支付类型创建对应的支付对象"""
if pay_type == "alipay":
return Alipay()
elif pay_type == "wechat":
return WeChatPay()
elif pay_type == "bank":
return BankPay()
else:
raise ValueError(f"不支持的支付类型：{pay_type}")

# 使用工厂创建对象
payment1 = PaymentFactory.create_payment("alipay")
print(payment1.pay(100)) # 输出：使用支付宝支付100元

payment2 = PaymentFactory.create_payment("wechat")
print(payment2.pay(200)) # 输出：使用微信支付200元

3. 装饰器模式：动态扩展对象功能

装饰器模式用于在不修改原对象的前提下，动态为对象添加额外功能，与 Python 装饰器语法异曲同工，但更侧重对象层面的扩展。

# 基础组件：咖啡
class Coffee:
def cost(self) -> float:
"""返回咖啡价格"""
return 10.0

def description(self) -> str:
"""返回咖啡描述"""
return "原味咖啡"

# 装饰器基类（继承自Coffee，保持接口一致）
class CoffeeDecorator(Coffee):
def __init__(self, coffee: Coffee):
self.coffee = coffee

# 具体装饰器：加牛奶
class MilkDecorator(CoffeeDecorator):
def cost(self) -> float:
return self.coffee.cost() + 2.0

def description(self) -> str:
return f"{self.coffee.description()} + 牛奶"

# 具体装饰器：加糖
class SugarDecorator(CoffeeDecorator):
def cost(self) -> float:
return self.coffee.cost() + 1.0

def description(self) -> str:
return f"{self.coffee.description()} + 糖"

# 具体装饰器：加奶泡
class FoamDecorator(CoffeeDecorator):
def cost(self) -> float:
return self.coffee.cost() + 3.0

def description(self) -> str:
return f"{self.coffee.description()} + 奶泡"

# 使用装饰器模式扩展功能
# 原味咖啡
coffee = Coffee()
print(f"{coffee.description()}：{coffee.cost()}元") # 输出：原味咖啡：10.0元

# 加牛奶的咖啡
coffee_with_milk = MilkDecorator(coffee)
print(f"{coffee_with_milk.description()}：{coffee_with_milk.cost()}元") # 输出：原味咖啡 + 牛奶：12.0元

# 加牛奶+糖的咖啡
coffee_with_milk_sugar = SugarDecorator(coffee_with_milk)
print(f"{coffee_with_milk_sugar.description()}：{coffee_with_milk_sugar.cost()}元") # 输出：原味咖啡 + 牛奶 + 糖：13.0元

# 加牛奶+糖+奶泡的咖啡
coffee_full = FoamDecorator(coffee_with_milk_sugar)
print(f"{coffee_full.description()}：{coffee_full.cost()}元") # 输出：原味咖啡 + 牛奶 + 糖 + 奶泡：16.0元

五、并发编程：提升程序执行效率

Python 的并发编程主要包括多线程、多进程、异步编程，适用于不同的场景（IO 密集型 / CPU 密集型）。

1. 多线程：适用于 IO 密集型任务

多线程（Thread）是 Python 中处理 IO 密集型任务（如网络请求、文件读写）的首选，通过threading模块实现。注意：Python 的 GIL（全局解释器锁）导致多线程无法利用多核 CPU，CPU 密集型任务推荐用多进程。

import threading
import time
import requests
from typing import List

# 全局变量（线程共享）
result_list = []
# 线程锁：解决多线程共享变量的竞争问题
lock = threading.Lock()

def fetch_url(url: str) -> None:
"""线程任务：请求指定URL并保存结果"""
try:
response = requests.get(url, timeout=5)
# 加锁修改共享变量
with lock:
result_list.append({"url": url, "status": response.status_code})
print(f"线程{threading.current_thread().name}完成：{url}")
except Exception as e:
with lock:
result_list.append({"url": url, "error": str(e)})
print(f"线程{threading.current_thread().name}失败：{url} – {e}")

def main():
# 待请求的URL列表（IO密集型任务）
urls = [
"https://www.baidu.com",
"https://www.google.com",
"https://www.github.com",
"https://www.python.org"
]

# 创建线程列表
threads: List[threading.Thread] = []
start_time = time.time()

# 创建并启动线程
for i, url in enumerate(urls):
thread = threading.Thread(target=fetch_url, args=(url,), name=f"Thread-{i+1}")
threads.append(thread)
thread.start()

# 等待所有线程完成
for thread in threads:
thread.join()

end_time = time.time()
print(f"\\n所有线程执行完成，总耗时：{end_time – start_time:.2f}秒")
print("执行结果：")
for item in result_list:
print(item)

if __name__ == "__main__":
main()

2. 多进程：适用于 CPU 密集型任务

多进程（Process）通过multiprocessing模块实现，可利用多核 CPU，适合处理 CPU 密集型任务（如数据计算、图像处理）。

import multiprocessing
import time
from typing import List

def calculate_sum(start: int, end: int, queue: multiprocessing.Queue) -> None:
"""进程任务：计算start到end的累加和，将结果放入队列"""
total = sum(range(start, end + 1))
# 多进程间通信：通过队列传递结果（共享内存不推荐）
queue.put(total)
print(f"进程{multiprocessing.current_process().name}完成计算：{start}~{end} = {total}")

def main():
# 计算1~100000000的累加和（CPU密集型任务）
total_range = 100000000
# 分为4个进程处理
process_num = 4
step = total_range // process_num

# 创建队列用于进程间通信
queue = multiprocessing.Queue()
# 创建进程列表
processes: List[multiprocessing.Process] = []
start_time = time.time()

# 创建并启动进程
for i in range(process_num):
start = i * step + 1
end = (i + 1) * step if i != process_num – 1 else total_range
process = multiprocessing.Process(
target=calculate_sum,
args=(start, end, queue),
name=f"Process-{i+1}"
)
processes.append(process)
process.start()

# 等待所有进程完成
for process in processes:
process.join()

# 汇总结果
total_sum = 0
while not queue.empty():
total_sum += queue.get()

end_time = time.time()
print(f"\\n所有进程执行完成，总耗时：{end_time – start_time:.2f}秒")
print(f"1~{total_range}的累加和：{total_sum}")

if __name__ == "__main__":
# Windows系统必须加if __name__ == "__main__"，避免递归创建进程
multiprocessing.freeze_support()
main()

3. 异步编程：高效处理 IO 密集型任务

异步编程（Asyncio）是 Python 3.5 + 引入的协程（Coroutine）机制，通过单线程切换任务，比多线程更高效，适合高并发 IO 密集型任务（如网络爬虫、API 服务）。

import asyncio
import aiohttp
import time
from typing import List

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
"""异步任务：请求指定URL"""
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(5)) as response:
return {"url": url, "status": response.status}
except Exception as e:
return {"url": url, "error": str(e)}

async def main():
# 待请求的URL列表
urls = [
"https://www.baidu.com",
"https://www.github.com",
"https://www.python.org",
"https://www.stackoverflow.com"
]

start_time = time.time()

# 创建异步HTTP会话
async with aiohttp.ClientSession() as session:
# 创建任务列表
tasks: List[asyncio.Task] = [asyncio.create_task(fetch_url(session, url)) for url in urls]
# 等待所有任务完成
results = await asyncio.gather(*tasks)

end_time = time.time()
print(f"\\n所有异步任务执行完成，总耗时：{end_time – start_time:.2f}秒")
print("执行结果：")
for result in results:
print(result)

if __name__ == "__main__":
# 运行异步主函数
asyncio.run(main())