處理 GB 級甚至 TB 級文本文件 (日誌、數據導出、基因組序列) 時，**絕不可**一次性 `read()` 或 `readlines()` 載入記憶體。Python 的文件對象本身是迭代器，配合 `with` 語句上下文管理器，提供了最優雅、最高效、最安全的逐行處理模式。本文詳細說明標準做法、編碼處理、異常捕獲、進度顯示等生產級技巧。
def process_large_file(filepath: str, encoding: str = 'utf-8') -> None:
"""
逐行處理大文件，記憶體佔用 O(1) (僅單行緩衝)。
"""
try:
with open(filepath, 'r', encoding=encoding, errors='replace') as f:
for line_num, line in enumerate(f, 1):
# line 保留末尾換行符 '\n'，按需 strip()
line = line.rstrip('\n\r')
if not line: continue # 跳過空行
# === 在這裡處理單行數據 ===
process_line(line, line_num)
# 可選：進度報告
if line_num % 1_000_000 == 0:
print(f"已處理 {line_num:,} 行...")
except FileNotFoundError:
print(f"錯誤：文件不存在 - {filepath}")
except PermissionError:
print(f"錯誤：無讀取權限 - {filepath}")
except UnicodeDecodeError as e:
print(f"編碼錯誤 (第 {e.start} 位置): {e.reason}, 建議指定正確 encoding 參數")
except Exception as e:
print(f"未預期錯誤: {e}")
def process_line(line: str, line_num: int) -> None:
# 業務邏輯：解析、過濾、寫入資料庫、發送隊列等
pass| 參數 | 推薦值 | 說明 |
|---|---|---|
mode='r' | 'r' / 'rb' | 文本模式自動解碼；二進制模式返回 bytes，需手動 decode |
encoding | 'utf-8' / 'gbk' / 'latin-1' | **必須顯式指定**，避免平台預設編碼差異 (Windows 默認 cp1252/gbk，Linux 默認 utf-8) |
errors | 'replace' / 'ignore' / 'strict' | 'replace': 非法字節替換為 (推薦，不中斷)；'ignore': 直接丟棄；'strict': 拋異常 (默認) |
newline | '' (默認) / '\n' | '' 通用換行符模式：自動識別 `\n`, `\r`, `\r\n` 並統一轉為 `\n` |
with open(filepath, 'rb') as f:
for raw_line in f:
try:
line = raw_line.decode('utf-8').rstrip('\n\r')
except UnicodeDecodeError:
line = raw_line.decode('gbk', errors='replace').rstrip('\n\r') # 回退嘗試
process_line(line)def read_chunks(filepath: str, chunk_size: int = 1024*1024): # 1MB
with open(filepath, 'rb') as f:
while True:
chunk = f.read(chunk_size)
if not chunk: break
yield chunk
# 使用
for chunk in read_chunks('huge.dat'):
process_binary_chunk(chunk)from concurrent.futures import ProcessPoolExecutor
def process_chunk(lines: list[str]) -> list[Result]:
return [heavy_compute(line) for line in lines]
def parallel_process(filepath: str, workers: int = 4, batch_size: int = 10000):
with open(filepath, 'r', encoding='utf-8') as f, \
ProcessPoolExecutor(max_workers=workers) as executor:
batch = []
futures = []
for line in f:
batch.append(line.rstrip('\n'))
if len(batch) >= batch_size:
futures.append(executor.submit(process_chunk, batch))
batch = []
if batch: futures.append(executor.submit(process_chunk, batch))
for future in futures:
for result in future.result():
save_result(result)import mmap
with open(filepath, 'r+b') as f:
mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
# mm 行為類似 bytes 對象，支援切片、find、正則
# 但逐行迭代不如 open() 直觀
for line in iter(mm.readline, b''):
process_line(line.decode().rstrip())
mm.close()| 錯誤代碼 | 問題 | 正確做法 |
|---|---|---|
| `lines = f.readlines()` | 一次性載入全文件到列表，OOM 風險 | 直接 `for line in f:` 迭代 |
| `content = f.read()` | 一次性載入全文件到字符串，OOM 風險 | 流式處理或 mmap |
| `open('file.txt')` 無 encoding | 跨平台亂碼 (Windows GBK vs Linux UTF-8) | 顯式 `encoding='utf-8'` |
| 無 `with` 手動 `f.close()` | 異常時文件句柄洩漏 | 必須用 `with` 上下文管理器 |
| 二進制模式 `for line in f:` 直接處理 | `line` 是 `bytes`，非 `str`，正則/字符串操作報錯 | `.decode(encoding)` 後處理 |
Python 處理大文件的核心心法：**流式迭代 + 顯式編碼 + 上下文管理器 + 異常兜底**。`with open(...) as f: for line in f:` 這一行代碼蘊含了資源安全、記憶體常數級佔用、跨平台換行符統一、編碼錯誤容錯等工程智慧。對於 CPU 密集型後續處理，引入 `ProcessPoolExecutor` 批量分發；對於隨機訪問需求，使用 `mmap`。拒絕 `read()`/`readlines()`，從源頭消除 OOM 隱患。
