Python脚本如何异步执行数据同步任务

本文目录导读：

1. 使用 asyncio 原生异步方案
2. 使用 aiofiles 进行文件操作
3. 使用 concurrent.futures 实现线程池/进程池
4. 定时异步同步任务
5. 完整的错误处理和重试机制
6. 推荐实践方案

我来为您介绍几种在Python中异步执行数据同步任务的方法：

使用 asyncio 原生异步方案

基础示例

import asyncio
import aiohttp
import aiofiles
import json
from datetime import datetime
class AsyncDataSync:
    def __init__(self):
        self.sync_status = {}
    async def fetch_data(self, source_url):
        """异步获取数据"""
        async with aiohttp.ClientSession() as session:
            async with session.get(source_url) as response:
                if response.status == 200:
                    return await response.json()
                return None
    async def write_data(self, data, filename):
        """异步写入数据"""
        async with aiofiles.open(filename, 'w') as f:
            await f.write(json.dumps(data, indent=2))
    async def sync_task(self, source_url, target_file):
        """单个同步任务"""
        task_name = f"sync_{source_url.split('/')[-1]}"
        self.sync_status[task_name] = "running"
        try:
            print(f"开始同步: {source_url}")
            data = await self.fetch_data(source_url)
            if data:
                await self.write_data(data, target_file)
                self.sync_status[task_name] = "completed"
                print(f"同步完成: {target_file}")
            else:
                self.sync_status[task_name] = "failed"
                print(f"获取数据失败: {source_url}")
        except Exception as e:
            self.sync_status[task_name] = "error"
            print(f"同步出错: {e}")
async def main():
    sync = AsyncDataSync()
    # 定义多个同步任务
    tasks = [
        sync.sync_task("https://api.example1.com/data", "data1.json"),
        sync.sync_task("https://api.example2.com/data", "data2.json"),
        sync.sync_task("https://api.example3.com/data", "data3.json"),
    ]
    # 并行执行所有任务
    await asyncio.gather(*tasks)
    # 查看状态
    print("同步状态:", sync.sync_status)
# 执行
if __name__ == "__main__":
    asyncio.run(main())

使用 aiofiles 进行文件操作

import asyncio
import aiofiles
import csv
import json
from pathlib import Path
class FileSyncTask:
    async def read_csv(self, filepath):
        """异步读取CSV文件"""
        data = []
        async with aiofiles.open(filepath, mode='r', encoding='utf-8') as f:
            content = await f.read()
            reader = csv.DictReader(content.splitlines())
            for row in reader:
                data.append(row)
        return data
    async def process_data(self, data):
        """模拟数据处理（异步）"""
        await asyncio.sleep(1)  # 模拟耗时操作
        processed = []
        for item in data:
            # 进行数据处理
            item['processed_time'] = asyncio.get_running_loop().time()
            processed.append(item)
        return processed
    async def sync_to_db(self, data, db_path):
        """异步写入到新文件（模拟数据库）"""
        processed = await self.process_data(data)
        async with aiofiles.open(db_path, 'w') as f:
            await f.write(json.dumps(processed, indent=2))
async def batch_sync():
    task = FileSyncTask()
    # 批量处理多个文件
    files = ["data1.csv", "data2.csv", "data3.csv"]
    async def process_file(filename):
        data = await task.read_csv(filename)
        output = f"processed_{filename.replace('.csv', '.json')}"
        await task.sync_to_db(data, output)
        print(f"处理完成: {filename} -> {output}")
    # 并行处理所有文件
    await asyncio.gather(*[process_file(f) for f in files])

使用 concurrent.futures 实现线程池/进程池

import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import requests
import time
class HybridSyncExecutor:
    def __init__(self, max_workers=5):
        self.thread_executor = ThreadPoolExecutor(max_workers=max_workers)
        self.process_executor = ProcessPoolExecutor(max_workers=max_workers)
    def sync_data_blocking(self, url, output_file):
        """阻塞式同步操作"""
        try:
            response = requests.get(url, timeout=10)
            with open(output_file, 'w') as f:
                f.write(response.text)
            return f"成功: {url} -> {output_file}"
        except Exception as e:
            return f"失败: {url} - {str(e)}"
    async def run_async_tasks(self, sync_tasks):
        """异步运行多个同步任务"""
        loop = asyncio.get_running_loop()
        tasks = []
        for url, output in sync_tasks:
            # 在线程池中执行阻塞操作
            task = loop.run_in_executor(
                self.thread_executor,
                self.sync_data_blocking,
                url, output
            )
            tasks.append(task)
        # 并行执行所有任务
        results = await asyncio.gather(*tasks)
        return results
async def main_hybrid():
    executor = HybridSyncExecutor()
    sync_tasks = [
        ("https://api.example.com/data1", "output1.txt"),
        ("https://api.example.com/data2", "output2.txt"),
        ("https://api.example.com/data3", "output3.txt"),
    ]
    results = await executor.run_async_tasks(sync_tasks)
    for result in results:
        print(result)

定时异步同步任务

import asyncio
from datetime import datetime
import schedule
class ScheduledSync:
    def __init__(self):
        self.sync_jobs = []
    async def sync_job(self, name, func, *args):
        """异步同步任务"""
        try:
            print(f"[{datetime.now()}] 开始同步: {name}")
            await func(*args)
            print(f"[{datetime.now()}] 完成同步: {name}")
        except Exception as e:
            print(f"[{datetime.now()}] 同步失败 {name}: {e}")
    async def run_scheduled(self):
        """运行定时任务"""
        while True:
            current_time = datetime.now()
            # 检查是否需要执行任务
            for job in self.sync_jobs:
                if job['next_run'] <= current_time:
                    asyncio.create_task(self.sync_job(
                        job['name'], 
                        job['func'], 
                        *job.get('args', [])
                    ))
                    job['next_run'] = current_time + job['interval']
            await asyncio.sleep(60)  # 每分钟检查一次
    def add_job(self, name, func, interval, *args):
        """添加定时任务"""
        self.sync_jobs.append({
            'name': name,
            'func': func,
            'interval': interval,
            'args': args,
            'next_run': datetime.now()
        })
# 使用示例
async def sync_database():
    """模拟数据库同步"""
    await asyncio.sleep(5)  # 模拟同步耗时
    print("数据库同步完成")
async def sync_files():
    """模拟文件同步"""
    await asyncio.sleep(3)
    print("文件同步完成")
async def main_scheduled():
    scheduler = ScheduledSync()
    # 添加定时任务
    scheduler.add_job("数据库同步", sync_database, 300)  # 每5分钟
    scheduler.add_job("文件同步", sync_files, 600)       # 每10分钟
    await scheduler.run_scheduled()

完整的错误处理和重试机制

import asyncio
import random
from typing import Callable, Any
class AsyncDataSyncWithRetry:
    def __init__(self, max_retries=3, retry_delay=1):
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    async def sync_with_retry(self, sync_func: Callable, *args, **kwargs) -> Any:
        """带重试的异步同步"""
        for attempt in range(self.max_retries):
            try:
                result = await sync_func(*args, **kwargs)
                return result
            except Exception as e:
                if attempt < self.max_retries - 1:
                    wait_time = self.retry_delay * (2 ** attempt)  # 指数退避
                    print(f"同步失败，{wait_time}秒后重试 (尝试 {attempt + 1}/{self.max_retries})")
                    await asyncio.sleep(wait_time)
                else:
                    raise e
    async def sync_all_with_error_handling(self, sync_tasks: list):
        """批量同步，带错误处理"""
        results = []
        for task in sync_tasks:
            try:
                result = await self.sync_with_retry(task['func'], *task.get('args', []))
                results.append({
                    'task': task['name'],
                    'status': 'success',
                    'data': result
                })
            except Exception as e:
                results.append({
                    'task': task['name'],
                    'status': 'failed',
                    'error': str(e)
                })
                # 记录错误日志
                print(f"任务 {task['name']} 最终失败: {e}")
        return results
# 使用示例
async def main_with_retry():
    sync = AsyncDataSyncWithRetry(max_retries=3, retry_delay=2)
    async def sync_user_data(user_id):
        # 模拟随机失败
        if random.random() < 0.5:
            raise Exception("网络错误")
        return {"user_id": user_id, "status": "synced"}
    tasks = [
        {'name': f'用户{i}数据同步', 'func': sync_user_data, 'args': (i,)}
        for i in range(5)
    ]
    results = await sync.sync_all_with_error_handling(tasks)
    print("同步结果:", results)

推荐实践方案

对于生产环境,建议：

1. 使用任务队列：Celery + Redis/RabbitMQ
2. 使用专门的异步框架：FastAPI + asyncio
3. 结合数据库：使用异步数据库驱动（asyncpg, aiomysql）
4. 监控和日志：集成日志系统，监控同步状态

选择哪种方案取决于您的具体需求：

• 简单任务：asyncio + aiohttp
• 混合同步/异步：concurrent.futures
• 定时任务：schedule + asyncio
• 生产环境：Celery 或 FastAPI 框架