后端性能优化实践笔记

作者: 醉月思发布日期: 2021-06-20 分类: 技术实践

最近把后端性能优化又过了一遍，留下几条实际有用的判断。

性能评估

基准测试

import time
import statistics
from functools import wraps

def benchmark(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        times = []
        
        # 运行 10 次
        for _ in range(10):
            start = time.time()
            result = func(*args, **kwargs)
            end = time.time()
            times.append(end - start)
        
        # 统计性能
        avg_time = statistics.mean(times)
        median_time = statistics.median(times)
        std_dev = statistics.stdev(times)
        
        print(f"{func.__name__}:")
        print(f"  Average: {avg_time:.4f}s")
        print(f"  Median: {median_time:.4f}s")
        print(f"  Std Dev: {std_dev:.4f}s")
        
        return result
    
    return wrapper

# 使用
@benchmark
def process_data(data):
    result = []
    for item in data:
        processed = item * 2
        result.append(processed)
    return result

data = list(range(10000))
process_data(data)

性能分析

import cProfile
import pstats

def profile_function(func):
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        result = profiler.runcall(func, *args, **kwargs)
        
        stats = pstats.Stats(profiler)
        stats.sort_stats('cumulative')
        stats.print_stats(10)  # 打印前 10 个最耗时的函数
        
        return result
    return wrapper

# 使用
@profile_function
def complex_operation(data):
    for item in data:
        time.sleep(0.001)
        return item * 2

complex_operation(data)

架构优化

水平扩展

# 使用负载均衡
from flask import Flask
import requests

app = Flask(__name__)

# 后端服务器列表
BACKEND_SERVERS = [
    'http://server1:8000',
    'http://server2:8000',
    'http://server3:8000'
]

current_server = 0

@app.route('/api/data')
def get_data():
    global current_server
    
    # 简单的轮询负载均衡
    server = BACKEND_SERVERS[current_server]
    current_server = (current_server + 1) % len(BACKEND_SERVERS)
    
    # 转发请求
    response = requests.get(f"{server}/data")
    return response.json()

# 或使用专业的负载均衡器（Nginx、HAProxy）

缓存层

from functools import lru_cache
import redis

# 内存缓存
@lru_cache(maxsize=1000)
def expensive_calculation(x):
    time.sleep(0.1)  # 模拟耗时操作
    return x * x

# Redis 缓存
class RedisCache:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def get(self, key):
        cached = self.redis.get(key)
        if cached:
            return cached.decode()
        return None
    
    def set(self, key, value, ttl=3600):
        self.redis.setex(key, ttl, value)
    
    def get_or_set(self, key, ttl=3600):
        cached = self.get(key)
        if cached:
            return cached
        
        # 计算值
        value = calculate_value(key)
        
        # 缓存结果
        self.set(key, value, ttl)
        
        return value

# 使用
redis_client = redis.Redis(host='localhost', port=6379, db=0)
cache = RedisCache(redis_client)

def cached_expensive_calculation(x):
    return cache.get_or_set(f"calc_{x}")

异步处理

from celery import Celery
import time

# 配置 Celery
app = Celery('tasks', broker='redis://localhost:6379/0')

# 定义异步任务
@app.task
def async_expensive_task(data):
    time.sleep(5)  # 模拟耗时操作
    return process_data(data)

# 调用异步任务
def process_request(data):
    # 立即返回
    task = async_expensive_task.delay(data)
    return {"task_id": task.id, "status": "processing"}

# 查询任务状态
def get_task_status(task_id):
    task = app.AsyncResult(task_id)
    
    if task.ready():
        return {
            "status": "completed",
            "result": task.result
        }
    else:
        return {"status": "processing"}

代码优化

算法优化

# 优化前：O(n²)
def find_duplicates_slow(arr):
    duplicates = []
    for i in range(len(arr)):
        for j in range(i + 1, len(arr)):
            if arr[i] == arr[j] and arr[i] not in duplicates:
                duplicates.append(arr[i])
    return duplicates

# 优化后：O(n)
def find_duplicates_fast(arr):
    seen = set()
    duplicates = set()
    
    for item in arr:
        if item in seen:
            duplicates.add(item)
        else:
            seen.add(item)
    
    return list(duplicates)

# 性能对比
import timeit

arr = list(range(10000)) + list(range(1000))

print("Slow:", timeit.timeit(lambda: find_duplicates_slow(arr), number=100))
print("Fast:", timeit.timeit(lambda: find_duplicates_fast(arr), number=100))

数据结构优化

# 使用更高效的数据结构
from collections import defaultdict

# 优化前：嵌套列表查找
def get_user_by_id_slow(users, user_id):
    for user in users:
        if user['id'] == user_id:
            return user
    return None

# 优化后：字典查找
def build_user_map(users):
    user_map = {}
    for user in users:
        user_map[user['id']] = user
    return user_map

def get_user_by_id_fast(user_map, user_id):
    return user_map.get(user_id)

# 使用示例
users = [
    {'id': 1, 'name': 'Alice'},
    {'id': 2, 'name': 'Bob'},
    {'id': 3, 'name': 'Charlie'}
]

user_map = build_user_map(users)
user = get_user_by_id_fast(user_map, 2)  # O(1) 查找

内存优化

# 使用生成器减少内存占用
def process_large_file(filename):
    with open(filename) as f:
        for line in f:
            # 逐行处理，不需要将整个文件加载到内存
            yield process_line(line)

# 使用 __slots__ 减少对象内存占用
class OptimizedUser:
    __slots__ = ['id', 'name', 'email']
    
    def __init__(self, id, name, email):
        self.id = id
        self.name = name
        self.email = email

# 批量处理
def batch_process(items, batch_size=1000):
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        process_batch(batch)

数据库优化

查询优化

# 优化前：N+1 查询
def get_users_with_orders_slow(user_ids):
    users = []
    for user_id in user_ids:
        user = db.query(User).get(user_id)
        orders = db.query(Order).filter(Order.user_id == user_id).all()
        users.append({
            'user': user,
            'orders': orders
        })
    return users

# 优化后：使用 JOIN
def get_users_with_orders_fast(user_ids):
    users = db.query(User).join(Order).filter(
        User.id.in_(user_ids)
    ).all()
    
    result = {}
    for user in users:
        if user.id not in result:
            result[user.id] = {
                'user': user,
                'orders': []
            }
        result[user.id]['orders'].append(user)
    
    return list(result.values())

批量操作

# 批量插入
def batch_insert_users(users):
    batch_size = 1000
    for i in range(0, len(users), batch_size):
        batch = users[i:i+batch_size]
        db.bulk_insert_mappings(User, batch)
        db.commit()

# 批量更新
def batch_update_users(updates):
    batch_size = 1000
    for i in range(0, len(updates), batch_size):
        batch = updates[i:i+batch_size]
        for update in batch:
            db.query(User).filter(User.id == update['id']).update(update)
        db.commit()

踩过的坑

坑一：过早优化

一开始就优化所有代码，结果花了很多时间在不重要的地方。

解决：先找到瓶颈，再针对性优化。

# 性能分析后再优化
def profile_and_optimize():
    # 先分析性能
    profiler = cProfile.Profile()
    profiler.runcall(process_data, large_dataset)
    
    # 找出最慢的部分
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    
    # 只优化最慢的部分
    top_functions = stats.stats[:5]
    for func_name, stats_data in top_functions:
        optimize_function(func_name)

坑二：缓存失效

缓存失效后性能反而更差。

解决：合理的缓存策略和监控。

class SmartCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.stats = defaultdict(int)
    
    def get(self, key):
        self.stats['get'] += 1
        
        cached = self.redis.get(key)
        if cached:
            self.stats['hit'] += 1
            return cached.decode()
        
        self.stats['miss'] += 1
        return None
    
    def set(self, key, value, ttl=3600):
        self.redis.setex(key, ttl, value)
    
    def get_hit_rate(self):
        total = self.stats['get']
        if total == 0:
            return 0.0
        return self.stats['hit'] / total
    
    def adjust_ttl(self, hit_rate):
        # 根据命中率调整 TTL
        if hit_rate > 0.8:
            return 3600  # 高命中率，长 TTL
        elif hit_rate > 0.5:
            return 1800  # 中等命中率，中等 TTL
        else:
            return 300   # 低命中率，短 TTL

坑三：并发问题

多线程/多进程导致的数据竞争。

解决：使用锁或线程安全的数据结构。

import threading
from queue import Queue

# 线程安全的队列
task_queue = Queue()

# 使用线程锁
counter_lock = threading.Lock()
counter = 0

def safe_increment():
    global counter
    with counter_lock:
        counter += 1

# 使用线程安全的数据结构
from threading import Lock

class ThreadSafeDict:
    def __init__(self):
        self.dict = {}
        self.lock = Lock()
    
    def get(self, key):
        with self.lock:
            return self.dict.get(key)
    
    def set(self, key, value):
        with self.lock:
            self.dict[key] = value
    
    def delete(self, key):
        with self.lock:
            if key in self.dict:
                del self.dict[key]

性能监控

实时监控

from prometheus_client import Counter, Histogram, start_http_server

# 定义指标
request_count = Counter('requests_total', 'Total requests', ['method', 'endpoint'])
request_latency = Histogram('request_latency_seconds', 'Request latency')

# 中间件
@app.before_request
def before_request():
    request.start_time = time.time()

@app.after_request
def after_request(response):
    # 记录请求计数
    request_count.labels(
        method=request.method,
        endpoint=request.endpoint
    ).inc()
    
    # 记录延迟
    if hasattr(request, 'start_time'):
        latency = time.time() - request.start_time
        request_latency.observe(latency)
    
    return response

# 启动监控服务
start_http_server(8001)

告警配置

# 配置告警规则
alert_rules = {
    'high_latency': {
        'threshold': 0.5,  # 秒
        'window': '5m',
        'action': 'send_alert'
    },
    'error_rate': {
        'threshold': 0.05,  # 5%
        'window': '1m',
        'action': 'restart_service'
    }
}

def check_alerts():
    current_latency = get_current_latency()
    current_error_rate = get_current_error_rate()
    
    if current_latency > alert_rules['high_latency']['threshold']:
        send_alert('High latency detected')
    
    if current_error_rate > alert_rules['error_rate']['threshold']:
        restart_service()

# 定期检查
import schedule
schedule.every(1).minutes.do(check_alerts)

写在最后

后端性能优化这东西，不只是技术，是系统思维。

优化了：

响应时间
吞吐量
资源利用率

带来了：

复杂度增加
维护成本
学习成本

优化之前先评估：

性能指标
用户体验
团队能力
预算

不是所有代码都需要优化，有时候增加服务器更简单。

这次后端性能优化花了一个月，从架构到代码。优化完成后，响应时间减少了 70%，吞吐量提升了 3 倍，用户体验明显改善。

可用性说明：本文发布于 2021 年 6 月，距今已超过五年。文中涉及的软件版本、接口、下载地址、命令参数和操作界面可能已经发生变化，部分方案在当前环境下可能失效。请结合官方最新文档核对后再操作，生产环境使用前务必先行验证。

版权声明: 本文首发于指尖魔法屋-后端性能优化实践笔记（https://blog.thinkmoon.cn/post/94-backend-performance-optimization-architecture-code-guide/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

用 GolemBot 把 Codex 接到飞书机器人

这篇文章记录一种少写代码的方案：用 GolemBot 接飞书机器人，用 Codex 做 Agent，再通过飞书 CLI Skills 读取文档、需求清单和 bug 清单。

📁 技术实践

📅 2026-06-09

调试 OpenAI Responses API 的 Web Search 踩坑记录

本文记录一次在投资研究 Agent 中调试 OpenAI Responses API Web Search 工具的过程，重点梳理请求格式、流式 SSE 解析、工具名兼容性和结构化数据源设计。

📁 技术实践

📅 2026-04-30

后端性能优化实践笔记

性能评估

基准测试

性能分析

架构优化

水平扩展

缓存层

异步处理

代码优化

算法优化

数据结构优化

内存优化

数据库优化

查询优化

批量操作

踩过的坑

坑一：过早优化

坑二：缓存失效

坑三：并发问题

性能监控

实时监控

告警配置

写在最后

评论

相关阅读