利用日志记录与性能分析工具优化Python应用程序效率

import logging

# 配置日志记录
logging.basicConfig(
    level=logging.DEBUG,  # 设置日志级别
    format='%(asctime)s - %(levelname)s - %(message)s',  # 设置日志格式
    handlers=[logging.StreamHandler()]  # 设置日志输出方式
)

# 记录一些日志
logging.debug('调试信息')
logging.info('常规信息')
logging.warning('警告信息')
logging.error('错误信息')
logging.critical('严重错误')

import time

def heavy_computation(n):
    start_time = time.time()  # 记录函数开始时间
    logging.info(f"Starting heavy computation with n={n}")

    # 模拟计算过程
    result = sum(i * i for i in range(n))

    end_time = time.time()  # 记录函数结束时间
    logging.info(f"Computation finished in {end_time - start_time:.4f} seconds")

    return result

# 调用函数进行测试
heavy_computation(1000000)

# 使用缓存来减少计算
cache = {}

def cached_computation(n):
    if n in cache:
        logging.info(f"Using cached result for n={n}")
        return cache[n]
    
    logging.info(f"Computing result for n={n}")
    result = sum(i * i for i in range(n))
    cache[n] = result
    return result

# 调用测试
cached_computation(1000000)
cached_computation(1000000)  # 第二次调用会使用缓存

# 使用数据库查询缓存来减少重复查询
import sqlite3

def get_user_data(user_id):
    query = f"SELECT * FROM users WHERE user_id = {user_id}"
    
    # 使用缓存来避免重复查询
    if user_id in data_cache:
        return data_cache[user_id]
    
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()
    cursor.execute(query)
    result = cursor.fetchall()
    
    data_cache[user_id] = result
    conn.close()
    
    return result

# 数据缓存
data_cache = {}

import asyncio

async def heavy_task(n):
    logging.info(f"Start heavy task with n={n}")
    await asyncio.sleep(3)  # 模拟耗时操作
    logging.info(f"Heavy task with n={n} finished")

async def main():
    await asyncio.gather(
        heavy_task(1),
        heavy_task(2),
        heavy_task(3)
    )

# 运行异步任务
asyncio.run(main())

import cProfile

def heavy_computation(n):
    result = sum(i * i for i in range(n))
    return result

# 使用cProfile来分析函数性能
cProfile.run('heavy_computation(1000000)')

         4 function calls in 0.138 seconds

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
        1    0.000    0.000    0.138    0.138 script.py:3(heavy_computation)
        1    0.000    0.000    0.000    0.000 {built-in method builtins.sum}
        1    0.138    0.138    0.138    0.138 {method 'disable' of '_lsprof.Profiler' objects}
        1    0.000    0.000    0.000    0.000 {built-in method builtins.exec}

def risky_operation():
    try:
        # 假设这是一个可能引发异常的操作
        result = 10 / 0  # 除零异常
    except Exception as e:
        logging.error(f"Exception occurred: {e}")
        raise  # 重新抛出异常

import tracemalloc

def memory_intensive_function():
    # 模拟一个内存占用大的操作
    large_list = [i * i for i in range(1000000)]

# 启动内存追踪
tracemalloc.start()

# 调用内存密集型函数
memory_intensive_function()

# 获取当前内存分配情况
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

# 打印出内存使用情况
for stat in top_stats[:10]:
    print(stat)

import concurrent.futures

def task(n):
    return sum(i * i for i in range(n))

# 使用ThreadPoolExecutor进行并发计算
with concurrent.futures.ThreadPoolExecutor() as executor:
    results = list(executor.map(task, [1000000, 2000000, 3000000]))

print(results)

# math_operations.pyx
def sum_of_squares(int n):
    cdef int i
    cdef long result = 0
    for i in range(n):
        result += i * i
    return result

$ cythonize -i math_operations.pyx

# 使用字典进行快速查找
cache = {}

def cached_computation(n):
    if n in cache:
        return cache[n]
    result = sum(i * i for i in range(n))
    cache[n] = result
    return result