四、并发编程

4.1 并发模型选择指南

4.2 多线程编程

import threading
import time
from concurrent.futures import ThreadPoolExecutor

# 创建线程
def worker(name, delay):
    print(f"线程{name}开始")
    time.sleep(delay)
    print(f"线程{name}结束")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i, 2))
    threads.append(t)
    t.start()

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

# 使用线程池
def fetch_url(url):
    import requests
    response = requests.get(url)
    return response.status_code

with ThreadPoolExecutor(max_workers=10) as executor:
    urls = ["https://example.com"] * 20
    results = executor.map(fetch_url, urls)
    for result in results:
        print(result)

4.3 多进程编程

from multiprocessing import Pool, cpu_count
import time

def cpu_bound_task(n):
    """CPU密集型任务：计算素数"""
    count = 0
    for i in range(2, n):
        if all(i % j != 0 for j in range(2, int(i**0.5) + 1)):
            count += 1
    return count

if __name__ == "__main__":
    numbers = [100000, 200000, 300000, 400000]

    # 使用进程池
    with Pool(processes=cpu_count()) as pool:
        start = time.time()
        results = pool.map(cpu_bound_task, numbers)
        print(f"结果：{results}")
        print(f"耗时：{time.time() - start:.2f}秒")

    # 对比单进程
    start = time.time()
    results = [cpu_bound_task(n) for n in numbers]
    print(f"单进程耗时：{time.time() - start:.2f}秒")

4.4 异步IO（asyncio）

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """异步获取URL"""
    async with session.get(url) as response:
        return await response.text()

async def fetch_all(urls):
    """并发获取多个URL"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# 运行异步任务
async def main():
    urls = ["https://httpbin.org/get"] * 10
    start = time.time()
    results = await fetch_all(urls)
    print(f"获取{len(results)}个页面，耗时：{time.time() - start:.2f}秒")

# Python 3.7+
asyncio.run(main())

# 异步生成器
async def async_counter(limit):
    for i in range(limit):
        await asyncio.sleep(1)
        yield i

async def consume():
    async for value in async_counter(5):
        print(value)

五、性能优化

5.1 代码级优化

# 1. 使用局部变量加速
def slow_loop():
    """慢：频繁访问全局变量"""
    for i in range(1000000):
        result = len("hello")  # 全局函数查找

def fast_loop():
    """快：局部变量缓存"""
    local_len = len  # 局部缓存
    for i in range(1000000):
        result = local_len("hello")

# 2. 列表推导 vs for循环
# 慢
squares = []
for i in range(1000):
    squares.append(i**2)

# 快
squares = [i**2 for i in range(1000)]

# 3. 使用join拼接字符串
# 慢
result = ""
for s in strings:
    result += s  # 每次创建新字符串

# 快
result = "".join(strings)

# 4. 使用集合进行成员检查
# 慢：O(n)
items = [1, 2, 3, 4, 5]
if 3 in items:  # 列表查找O(n)
    pass

# 快：O(1)
items_set = {1, 2, 3, 4, 5}
if 3 in items_set:  # 集合查找O(1)
    pass

5.2 内存优化

import sys
import array

# 1. 使用__slots__减少内存占用
class RegularUser:
    def __init__(self, name, age):
        self.name = name
        self.age = age

class SlimUser:
    __slots__ = ('name', 'age')
    def __init__(self, name, age):
        self.name = name
        self.age = age

regular = RegularUser("张三", 25)
slim = SlimUser("李四", 30)
print(f"常规类内存：{sys.getsizeof(regular.__dict__)} 字节")
print(f"__slots__类内存：{sys.getsizeof(slim)} 字节")

# 2. 使用生成器代替列表
def read_large_file(file_path):
    """生成器逐行读取，不一次性加载到内存"""
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()

# 3. 使用array模块存储同类型数据
import array
numbers = array.array('i', range(1000000))  # 比list省内存

5.3 使用C扩展加速

# 使用numpy加速数值计算
import numpy as np
import time

# Python原生：慢
start = time.time()
result = [i**2 for i in range(10000000)]
print(f"原生Python：{time.time() - start:.2f}秒")

# NumPy：快
start = time.time()
arr = np.arange(10000000)
result = arr ** 2
print(f"NumPy：{time.time() - start:.2f}秒")

# 使用Cython、Numba或PyPy进一步加速
from numba import jit

@jit(nopython=True)
def fast_sum(arr):
    total = 0
    for x in arr:
        total += x
    return total

六、工程化实践

6.1 包管理与虚拟环境

# 创建虚拟环境
python -m venv venv

# 激活环境
# Windows
venv\Scripts\activate
# Mac/Linux
source venv/bin/activate

# 管理依赖
pip install requests pandas
pip freeze > requirements.txt

# 从requirements.txt安装
pip install -r requirements.txt

# 使用poetry管理项目
poetry new myproject
poetry add requests
poetry install

6.2 日志系统

import logging
import logging.config

# 配置日志
LOGGING_CONFIG = {
    'version': 1,
    'formatters': {
        'standard': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        },
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'standard',
            'level': 'INFO'
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': 'app.log',
            'formatter': 'detailed',
            'level': 'DEBUG'
        }
    },
    'root': {
        'handlers': ['console', 'file'],
        'level': 'DEBUG'
    }
}

logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger(__name__)

# 使用
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.exception("捕获异常信息")

6.3 单元测试

import unittest
from unittest.mock import Mock, patch

# 待测试的函数
def add(a, b):
    return a + b

def get_data_from_api(url):
    import requests
    response = requests.get(url)
    return response.json()

# 测试类
class TestMathFunctions(unittest.TestCase):
    def test_add_positive(self):
        self.assertEqual(add(1, 2), 3)

    def test_add_negative(self):
        self.assertEqual(add(-1, -2), -3)

    def test_add_zero(self):
        self.assertEqual(add(5, 0), 5)

class TestApiCall(unittest.TestCase):
    @patch('requests.get')
    def test_get_data_from_api(self, mock_get):
        # 模拟API响应
        mock_response = Mock()
        mock_response.json.return_value = {"data": "test"}
        mock_get.return_value = mock_response

        result = get_data_from_api("https://api.example.com")
        self.assertEqual(result, {"data": "test"})
        mock_get.assert_called_once_with("https://api.example.com")

# 使用pytest运行测试
# pytest test_module.py

七、设计模式：可复用解决方案
7.1 单例模式

class Singleton:
    _instance = None

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

    def __init__(self, name):
        # 注意：__init__每次都会被调用
        if not hasattr(self, 'initialized'):
            self.name = name
            self.initialized = True

# 使用模块实现单例（更Pythonic）
# config.py
class Config:
    def __init__(self):
        self.settings = {}

config = Config()  # 模块级单例

7.2 工厂模式

from abc import ABC, abstractmethod

class PaymentMethod(ABC):
    @abstractmethod
    def pay(self, amount):
        pass

class CreditCard(PaymentMethod):
    def pay(self, amount):
        print(f"信用卡支付：{amount}元")

class PayPal(PaymentMethod):
    def pay(self, amount):
        print(f"PayPal支付：{amount}元")

class Alipay(PaymentMethod):
    def pay(self, amount):
        print(f"支付宝支付：{amount}元")

class PaymentFactory:
    _methods = {
        'credit': CreditCard,
        'paypal': PayPal,
        'alipay': Alipay
    }

    @classmethod
    def create(cls, method_type):
        payment_class = cls._methods.get(method_type)
        if not payment_class:
            raise ValueError(f"不支持的支付方式：{method_type}")
        return payment_class()

# 使用
payment = PaymentFactory.create('alipay')
payment.pay(100)

Python进阶之路是一场持续的修行，从面向对象进阶到高级特性，从函数式编程到并发模型，每一步都在拓展你的编程思维。进阶的关键不在于记忆更多的语法，而在于建立系统性思维——能够选择最合适的工具解决具体问题，能够写出可维护、可测试、高性能的代码。
来源：
https://app-adtysnu98v0h.appmiaoda.com