Python脚本如何优化海量数据同步吞吐

wen python案例 2026-06-25 270

本文目录导读：

核心IO模型优化
数据库连接优化
数据流处理优化
网络传输优化
并行处理框架选择
具体优化实践案例
关键性能调优参数
监控与调优工具
最佳实践建议

针对Python海量数据同步的吞吐优化，需要从IO模型、并行处理、内存管理、数据压缩、协议选择等多个维度进行,以下是经过实践验证的优化策略：

核心IO模型优化

异步IO（首选方案）

import asyncio
import aiohttp
import aiofiles
async def batch_fetch(urls, batch_size=1000):
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=30)
    async with aiohttp.ClientSession(connector=connector) as session:
        semaphore = asyncio.Semaphore(200)
        tasks = []
        for url in urls:
            tasks.append(fetch_with_limit(session, url, semaphore))
        return await asyncio.gather(*tasks)
async def fetch_with_limit(session, url, semaphore):
    async with semaphore:
        async with session.get(url) as resp:
            return await resp.json()

多进程 + 协程（CPU密集型）

from multiprocessing import Pool
import asyncio
def worker_process(chunk_data):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    return loop.run_until_complete(async_processor(chunk_data))

数据库连接优化

连接池配置

# PostgreSQL 优化
import asyncpg
pool = await asyncpg.create_pool(
    min_size=10,
    max_size=50,
    max_queries=10000,
    max_inactive_connection_lifetime=300,
    command_timeout=60
)
# MySQL 优化
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
    "mysql+asyncmy://user:pass@host/db",
    pool_size=20,
    max_overflow=40,
    pool_pre_ping=True,
    echo=False
)

批量操作

# 批量插入（每次1000-5000条）
async def bulk_insert(conn, records, batch_size=2000):
    for i in range(0, len(records), batch_size):
        batch = records[i:i+batch_size]
        await conn.executemany(
            "INSERT INTO table (col1, col2) VALUES ($1, $2)",
            batch
        )

数据流处理优化

使用生成器避免内存溢出

def chunked_reader(file_path, chunk_size=100000):
    """流式读取大文件"""
    with open(file_path, 'r') as f:
        chunk = []
        for line in f:
            chunk.append(line)
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
        if chunk:
            yield chunk

零拷贝数据处理

import io
import csv
def process_csv_stream(stream, batch_size=5000):
    """避免加载全部到内存"""
    reader = csv.DictReader(stream)
    batch = []
    for row in reader:
        batch.append(row)
        if len(batch) >= batch_size:
            yield batch
            batch = []
    if batch:
        yield batch

网络传输优化

压缩传输

import gzip
import requests
def compressed_fetch(url):
    headers = {'Accept-Encoding': 'gzip, deflate'}
    response = requests.get(url, headers=headers, stream=True)
    # 自动解压
    for chunk in response.iter_content(chunk_size=8192):
        yield chunk

连接复用

from urllib3 import PoolManager
http = PoolManager(
    num_pools=10,
    maxsize=100,
    block=False,
    retries=False
)
def batch_request(urls, method='GET'):
    with http.request(method, urls[0]) as resp:
        # 重用连接
        pass

并行处理框架选择

使用Python原生协程

import aiomultiprocess
import uvloop
async def main():
    uvloop.install()  # 加速事件循环
    async with aiomultiprocess.Pool(processes=4) as pool:
        results = await pool.map(process_chunk, data_chunks)

Ray分布式框架

import ray
ray.init(num_cpus=8)
@ray.remote
class DataProcessor:
    def process(self, data):
        # 高层封装，自动处理序列化
        pass

具体优化实践案例

优化前后对比

# 优化前：逐条处理
def slow_sync(source, target):
    for record in source:
        target.insert(record)  # 每条单独连接
# 优化后：批量+异步+连接池
async def fast_sync(source, target):
    async with target.pool.acquire() as conn:
        # 批量根据距离分批
        batch_size = 5000
        async for chunk in source.chunked_read(batch_size):
            await conn.executemany(
                "INSERT INTO target VALUES ($1, $2, $3)",
                chunk
            )

关键性能调优参数

参数	建议值	说明
批处理大小	1000-5000	太小增加网络开销，太大内存压力
连接池大小	CPU核心数×2-4	避免连接竞争
超时时间	30-60s	平衡重试频率
并发数	200-500	受限于系统文件描述符
缓冲区大小	64KB-1MB	网络IO优化

监控与调优工具

import time
import logging
from contextlib import contextmanager
@contextmanager
def measure_throughput(name):
    start = time.time()
    count = 0
    def tick():
        nonlocal count
        count += 1
    try:
        yield tick
    finally:
        elapsed = time.time() - start
        throughput = count / elapsed
        logging.info(f"{name}: {throughput:.0f} records/s")