异步批量处理¶
本页面介绍如何使用异步API进行批量处理。
批量创建文件¶
并发生成多个文件¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
async def create_file(filename, data):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("数据")
for idx, value in enumerate(data, start=1):
await sheet.set_cell_value(f"A{idx}", value)
await wb.save(filename)
async def batch_create():
files = [
("file1.xlsx", [1, 2, 3]),
("file2.xlsx", [4, 5, 6]),
("file3.xlsx", [7, 8, 9]),
]
tasks = [create_file(name, data) for name, data in files]
await asyncio.gather(*tasks)
asyncio.run(batch_create())
大数据处理¶
异步写入大量数据¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
async def write_large_dataset():
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("大数据")
chunk_size = 1000
total_rows = 10000
for start_row in range(1, total_rows, chunk_size):
tasks = []
for row in range(start_row, min(start_row + chunk_size, total_rows)):
task = sheet.set_cell_value(f"A{row}", f"Data {row}")
tasks.append(task)
await asyncio.gather(*tasks)
await wb.save("large_dataset.xlsx")
asyncio.run(write_large_dataset())
完整示例¶
示例: 批量报表生成¶
import asyncio
from datetime import date, timedelta
from symphra_excel.async_support import AsyncWorkbook
from symphra_excel.styles import PredefinedStyles
async def generate_daily_report(report_date, sales_data):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet(f"{report_date}")
await sheet.set_cell_value("A1", f"{report_date}销售报表")
await sheet.apply_style("A1", PredefinedStyles.title_style())
headers = ["产品", "数量", "销售额"]
for col_idx, header in enumerate(headers, start=1):
cell = f"{chr(64+col_idx)}2"
await sheet.set_cell_value(cell, header)
await sheet.apply_style(cell, PredefinedStyles.header_style())
row_tasks = []
for row_idx, item in enumerate(sales_data, start=3):
row_tasks.extend([
sheet.set_cell_value(f"A{row_idx}", item["product"]),
sheet.set_cell_value(f"B{row_idx}", item["quantity"]),
sheet.set_cell_value(f"C{row_idx}", item["amount"]),
])
await asyncio.gather(*row_tasks)
await wb.save(f"report_{report_date}.xlsx")
async def generate_monthly_reports():
start_date = date(2024, 1, 1)
days = 30
tasks = []
for day in range(days):
report_date = start_date + timedelta(days=day)
sales_data = [
{"product": "产品A", "quantity": 10, "amount": 1000},
{"product": "产品B", "quantity": 20, "amount": 2000},
]
task = generate_daily_report(report_date.isoformat(), sales_data)
tasks.append(task)
await asyncio.gather(*tasks)
asyncio.run(generate_monthly_reports())
性能优化¶
分批处理¶
async def batch_process(items, batch_size=10):
for i in range(0, len(items), batch_size):
batch = items[i:i+batch_size]
tasks = [process_item(item) for item in batch]
await asyncio.gather(*tasks)
async def process_item(item):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("数据")
await sheet.set_cell_value("A1", item)
await wb.save(f"output_{item}.xlsx")
最佳实践¶
✅ 推荐做法¶
async def efficient_batch():
batch_size = 100
for i in range(0, len(items), batch_size):
batch = items[i:i+batch_size]
tasks = [process(item) for item in batch]
await asyncio.gather(*tasks)
async with AsyncWorkbook() as wb:
await wb.save("output.xlsx")