Python批处理文件优化技巧和最佳实践

# 不推荐的写法：每次写入时都打开文件
for i in range(10000):
    with open('output.txt', 'a') as f:
        f.write(f"Line {i}\n")

# 推荐的写法：打开一次文件，进行多次写入
with open('output.txt', 'a') as f:
    for i in range(10000):
        f.write(f"Line {i}\n")

with open('input.txt', 'r') as f:
    for line in f:
        process(line)  # 每次读取处理一行数据

with open('input.txt', 'r') as f:
    lines = f.readlines()  # 一次性读取所有行
    for line in lines:
        process(line)

# 不推荐：一次性加载所有数据到内存
data = [process(i) for i in range(10000000)]

# 推荐：使用生成器，避免一次性加载所有数据到内存
def generate_data():
    for i in range(10000000):
        yield process(i)
        
for item in generate_data():
    pass

def process_large_file(file_path, chunk_size=1024):
    with open(file_path, 'r') as f:
        while chunk := f.read(chunk_size):
            process(chunk)  # 每次读取并处理一个数据块

# 不推荐：嵌套循环中过多的重复操作
for i in range(10000):
    for j in range(10000):
        if i == j:
            process(i)

# 推荐：将嵌套循环减少到最低，避免不必要的重复计算
for i in range(10000):
    process(i)  # 避免不必要的内层循环

# 不推荐：暴力算法进行排序
data = [9, 2, 5, 8, 7]
for i in range(len(data)):
    for j in range(i + 1, len(data)):
        if data[i] > data[j]:
            data[i], data[j] = data[j], data[i]  # 冒泡排序

# 推荐：使用内置的排序函数，它的时间复杂度为O(n log n)
data = [9, 2, 5, 8, 7]
data.sort()  # 内置排序更高效

from concurrent.futures import ThreadPoolExecutor

def process_data(data):
    # 处理每条数据
    pass

data = [1, 2, 3, 4, 5]
with ThreadPoolExecutor(max_workers=5) as executor:
    executor.map(process_data, data)

from multiprocessing import Pool

def process_data(data):
    # 处理每条数据
    pass

data = [1, 2, 3, 4, 5]
with Pool(processes=4) as pool:
    pool.map(process_data, data)

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ['http://example.com', 'http://example.org']
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

import logging

logging.basicConfig(level=logging.INFO)

for i in range(100000):
    logging.info(f"Processing item {i}")

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

for i in range(100000):
    if i % 1000 == 0:  # 每1000条记录一次日志
        logger.info(f"Processing item {i}")

for i in range(100):
    process_data(i)  # 如果process_data出现异常，整个任务将中断

for i in range(100):
    try:
        process_data(i)
    except Exception as e:
        logging.error(f"Error processing item {i}: {e}")  # 捕获异常并记录