本文目录导读：
我来为您介绍几种在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)
推荐实践方案
对于生产环境,建议：
- 使用任务队列：Celery + Redis/RabbitMQ
- 使用专门的异步框架：FastAPI + asyncio
- 结合数据库：使用异步数据库驱动（asyncpg, aiomysql）
- 监控和日志：集成日志系统，监控同步状态
选择哪种方案取决于您的具体需求：
- 简单任务：
asyncio+aiohttp - 混合同步/异步：
concurrent.futures - 定时任务：
schedule+asyncio - 生产环境：
Celery或FastAPI框架
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
