性能优化指南¶
本页面介绍如何优化Symphra Excel的性能,实现高效的Excel文件生成。
性能概览¶
Symphra Excel在不同规模下的典型性能表现:
| 数据规模 | 处理时间 | 内存占用 | 推荐方式 |
|---|---|---|---|
| <1,000行 | <1秒 | <50MB | Workbook |
| 1,000-10,000行 | 1-5秒 | 50-200MB | Workbook |
| 10,000-100,000行 | 5-30秒 | 100-500MB | MemoryOptimizedWorkbook |
| >100,000行 | 30秒+ | 500MB+ | MemoryOptimizedWorkbook + 分批 |
优化策略¶
1. 选择正确的工作簿类型¶
# ❌ 处理大数据时使用普通Workbook
from symphra_excel import Workbook
wb = Workbook()
for i in range(100000): # 可能导致内存溢出
sheet.set_cell_value(f"A{i}", i)
# ✅ 使用内存优化版本
from symphra_excel.memory_ops import MemoryOptimizedWorkbook
wb = MemoryOptimizedWorkbook()
for i in range(100000): # 内存占用稳定
sheet.set_cell_value(f"A{i}", i)
2. 样式复用¶
from symphra_excel import CellStyle
# ❌ 重复创建样式对象
for i in range(1000):
style = CellStyle() # 每次都创建新对象
style.set_font(bold=True)
sheet.apply_style(f"A{i}", style)
# ✅ 创建一次,多次使用
style = CellStyle()
style.set_font(bold=True)
for i in range(1000):
sheet.apply_style(f"A{i}", style)
3. 批量操作¶
# ❌ 逐个操作
for row in range(1000):
for col in range(10):
sheet.set_cell_value(f"{chr(65+col)}{row}", value)
sheet.apply_style(f"{chr(65+col)}{row}", style)
# ✅ 批量操作
data = [[value for _ in range(10)] for _ in range(1000)]
for row_idx, row_data in enumerate(data, start=1):
for col_idx, value in enumerate(row_data):
sheet.set_cell_value(f"{chr(65+col_idx)}{row_idx}", value)
# 批量应用样式
style = CellStyle()
for row in range(1000):
for col in range(10):
sheet.apply_style(f"{chr(65+col)}{row}", style)
4. 减少样式种类¶
# ❌ 每行使用不同颜色(太多样式)
from symphra_excel.styles import Color
colors = [Color.LIGHT_RED, Color.LIGHT_BLUE, ...]
for i, color in enumerate(colors):
style = CellStyle()
style.set_background_color(color)
sheet.apply_style(f"A{i}", style)
# ✅ 使用有限的样式集
header_style = CellStyle()
header_style.set_background_color(Color.DARK_BLUE)
data_style = CellStyle()
data_style.set_background_color(Color.WHITE)
# 只使用两种样式
sheet.apply_style("A1", header_style)
for i in range(2, 1000):
sheet.apply_style(f"A{i}", data_style)
5. 避免不必要的格式化¶
# ❌ 对所有单元格应用样式
for row in range(1, 1001):
for col in range(1, 11):
sheet.set_cell_value(f"{chr(64+col)}{row}", value)
sheet.apply_style(f"{chr(64+col)}{row}", style)
# ✅ 只对有数据的单元格应用样式
for row in range(1, 1001):
for col in range(1, 11):
if has_data(row, col):
sheet.set_cell_value(f"{chr(64+col)}{row}", value)
sheet.apply_style(f"{chr(64+col)}{row}", style)
6. 使用异步API处理并发¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
# ✅ 并发生成多个文件
async def create_files():
tasks = []
for i in range(10):
tasks.append(create_single_file(i))
await asyncio.gather(*tasks)
async def create_single_file(index):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("数据")
# ... 操作
await wb.save(f"file_{index}.xlsx")
asyncio.run(create_files())
内存优化技巧¶
1. 使用流式写入¶
from symphra_excel.memory_ops import MemoryOptimizedWorkbook
# 处理超大数据集
with MemoryOptimizedWorkbook() as wb:
sheet = wb.create_worksheet("大数据")
# 分批处理
batch_size = 10000
for batch_start in range(0, 1000000, batch_size):
for i in range(batch_size):
row = batch_start + i + 1
sheet.set_cell_value(f"A{row}", f"Data {row}")
# 定期刷新缓冲区
if batch_start % 100000 == 0:
print(f"已处理 {batch_start} 行")
wb.save("huge_file.xlsx")
2. 及时释放资源¶
# ✅ 使用上下文管理器
with Workbook() as wb:
# ... 操作
wb.save("output.xlsx")
# 自动释放资源
# ❌ 手动管理(容易忘记关闭)
wb = Workbook()
# ... 操作
wb.save("output.xlsx")
wb.close() # 容易忘记
3. 避免加载整个文件到内存¶
# ❌ 一次性加载大文件
wb = Workbook("large_file.xlsx") # 占用大量内存
# ... 处理
wb.save("modified.xlsx")
# ✅ 使用只读模式
wb = Workbook("large_file.xlsx", mode="read_only")
# 只读取需要的部分
value = wb.worksheets[0].get_cell_value("A1")
性能测试¶
基准测试代码¶
import time
from symphra_excel import Workbook
from symphra_excel.memory_ops import MemoryOptimizedWorkbook
def benchmark_standard(rows=10000):
start = time.time()
with Workbook() as wb:
sheet = wb.create_worksheet("测试")
for i in range(rows):
sheet.set_cell_value(f"A{i+1}", f"Data {i}")
wb.save("standard.xlsx")
elapsed = time.time() - start
print(f"标准模式: {rows}行, 耗时{elapsed:.2f}秒")
def benchmark_optimized(rows=10000):
start = time.time()
with MemoryOptimizedWorkbook() as wb:
sheet = wb.create_worksheet("测试")
for i in range(rows):
sheet.set_cell_value(f"A{i+1}", f"Data {i}")
wb.save("optimized.xlsx")
elapsed = time.time() - start
print(f"优化模式: {rows}行, 耗时{elapsed:.2f}秒")
# 运行测试
benchmark_standard(10000)
benchmark_optimized(10000)
内存监控¶
import tracemalloc
from symphra_excel import Workbook
# 开始内存监控
tracemalloc.start()
# 执行操作
with Workbook() as wb:
sheet = wb.create_worksheet("测试")
for i in range(10000):
sheet.set_cell_value(f"A{i+1}", f"Data {i}")
wb.save("output.xlsx")
# 获取内存使用情况
current, peak = tracemalloc.get_traced_memory()
print(f"当前内存: {current / 1024 / 1024:.2f} MB")
print(f"峰值内存: {peak / 1024 / 1024:.2f} MB")
tracemalloc.stop()
实战案例¶
案例1: 百万行数据导出¶
from symphra_excel.memory_ops import MemoryOptimizedWorkbook
from symphra_excel import CellStyle
# 创建样式(只创建一次)
header_style = CellStyle()
header_style.set_font(bold=True)
data_style = CellStyle()
data_style.set_number_format("#,##0.00")
with MemoryOptimizedWorkbook() as wb:
sheet = wb.create_worksheet("百万数据")
# 写入表头
headers = ["ID", "名称", "金额", "日期"]
for col_idx, header in enumerate(headers):
cell = f"{chr(65+col_idx)}1"
sheet.set_cell_value(cell, header)
sheet.apply_style(cell, header_style)
# 批量写入数据
batch_size = 10000
total_rows = 1000000
for batch_start in range(0, total_rows, batch_size):
batch_end = min(batch_start + batch_size, total_rows)
for i in range(batch_start, batch_end):
row = i + 2 # 从第2行开始(第1行是表头)
sheet.set_cell_value(f"A{row}", i + 1)
sheet.set_cell_value(f"B{row}", f"Item_{i+1}")
sheet.set_cell_value(f"C{row}", (i + 1) * 100)
sheet.set_cell_value(f"D{row}", "2025-01-01")
# 应用样式
sheet.apply_style(f"C{row}", data_style)
# 打印进度
if batch_end % 100000 == 0:
print(f"进度: {batch_end}/{total_rows} ({batch_end/total_rows*100:.1f}%)")
print("保存中...")
wb.save("million_rows.xlsx")
print("完成!")
案例2: 并发生成多个报表¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
async def create_report(dept_name, data):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet(dept_name)
# 写入数据
for row_idx, row_data in enumerate(data, start=1):
for col_idx, value in enumerate(row_data):
await sheet.set_cell_value(
f"{chr(65+col_idx)}{row_idx}",
value
)
await wb.save(f"{dept_name}_report.xlsx")
print(f"{dept_name} 报表已生成")
async def generate_all_reports():
departments = {
"销售部": [[1, 2, 3], [4, 5, 6]],
"技术部": [[7, 8, 9], [10, 11, 12]],
"人事部": [[13, 14, 15], [16, 17, 18]],
}
tasks = [
create_report(dept, data)
for dept, data in departments.items()
]
await asyncio.gather(*tasks)
print("所有报表生成完成!")
# 运行
asyncio.run(generate_all_reports())
性能检查清单¶
使用这个清单检查您的代码:
- [ ] 大数据使用了MemoryOptimizedWorkbook
- [ ] 样式对象被复用,没有重复创建
- [ ] 使用了批量操作而不是逐个处理
- [ ] 限制了样式的种类数量
- [ ] 只对需要的单元格应用样式
- [ ] 使用了上下文管理器管理资源
- [ ] 大文件使用了分批处理
- [ ] 并发场景使用了异步API
- [ ] 避免了不必要的文件加载
- [ ] 定期监控了内存使用情况