Django DRF 性能优化
1. DB 查询分析
步骤:
- 通过日志/监控等, 统计top 10的接口(URL及请求参数)
- 通过 APM 等, 统计慢接口top 10, 以及得到慢查询 SQL
- 开发环境, 开启SQL打印
django.db.backends, 可以参考Logging raw SQL to the console in Django - 构造
1/2步中收集的请求, 在开发环境复现 - 获取执行的 SQL
- 分析
分析:
- SQL 的查询条件是否命中索引, 是否有全表扫描(
mysql explain) - SQL 中查询返回字段是否可以减少
- 是否某些 SQL 不应该执行, 但是执行了(误用)
- 是否某些 SQL 可以合并成一次查询
- 是否可以不执行/只执行一次, 例如通过加入缓存
重点: 高频查询, 增加索引能解决很大一部分问题; 尽量避免全表扫描的存在.
2. serializer 优化
2.1 property or SerializerMethodField
N+1 selects problem
这两种, 在里面做了一些特殊操作, 例如执行一次db查询
class Foo(models.Model):
@property
def last_check_time(self):
# property 放大
cr = CheckRecords.objects.filter(is_success=True).latest()
return cr.check_time if cr or None
class FooOutputSLZ(serializers.ModelSerializer):
last_check_time = serializers.DateTimeField()
bar = serializers.SerializerMethodField()
def get_bar(self, obj):
return Bar.objects.filter(foo_type=obj.type).first()
此时, 列表查询, page_size=100, 会产生 1+100+100=201 次 db 查询
优化:
- 不返回
- 加缓存, 例如增加外部一个方法
get_bar_from_cache(foo_type: str) - 通过
serializer context在上层提前查好, 适合所有items公共的数据, 例如用户职位名称
class UserListApi(generics.ListAPIView):
def get_serializer_context(self):
# set into context, for slz to_representation
return {"position_name_map": get_position_name_map_from_db()}
class UserOutputSLZ(serializers.ModelSerializer):
position_name = serializers.SerializerMethodField()
def get_position_name(self, obj):
m = self.context.get("position_name_map")
return m.get(obj.position_id)
2.2 Serializer本身的性能问题
小改:
read_only=True- 自己构建数据, 不使用serializer
class UserOutputSLZ(serializers.ModelSerializer):
position = serializers.CharField(read_only=True)
data = []
for u in users:
data.append({
"id": user.id,
"name": user.name,
})
return Response(data=data, status=status.HTTP_200_OK)
3. queryset 优化
3.1 defer / only 等, 只查询想要的字段
如果模型中存在一些本次逻辑用不上的字段, 那么不要查询出来, 或者声明只查询想要的字段
- 例如
create_time/update_time description/content/extras等 TextField or JsonField
使用 django queryset only() 只获取指定字段
Department.objects.only("id", "name")
使用 django queryset defer() 延迟获取某些字段
Department.objects.defer("description", "extras", "create_time")
可以在打印的 SQL 中看到 SELECT 字段列表的变化.
3.2 values()/values_list()
以dict/tuple的形式获取结果列表, 避免对象创建
# got List[Dict]
Department.objects.values("id", "name")
# got List[Tuple]
Department.objects.values_list("id", "name")
values_list()如果只获取一个字段, 那么可以加上flat=True, 直接获取返回列表
Department.objects.values_list('id', flat=True)
3.3 prefetch_related / selected_related
N+1 selects problem
这两个的目的都是将可能的多次查询合并, 减少 DB 查询次数, 可以参考这篇文章 Django的 select_related 和 prefetch_related 函数对 QuerySet 查询的优化
- select_related(): “follow” foreign-key relationships, selecting additional related-object data when it executes its query. (SQL JOIN)
- prefetch_related(): automatically retrieve, in a single batch, related objects for each of the specified lookups.(ANOTHER SQL)
注意, select_related/prefetch_related 的表如果数据量特别大(百万), 或者单页page_size特别大(几千/上万), 可能导致查询直接卡死
3.4 其他
use count()
# bad
len(User.objects.all())
# good
User.objects.count()
use exists()
# bad
if User.objects.filter(username=x):
dosomething
# good
if User.objects.filter(username=x).exists():
dosomething
use bulk_create()/bulk_update()
Entry.objects.bulk_create(objs, 1000)
Entry.objects.bulk_update(objs, ['headline'], 1000)
specific the updated fields
product.name = 'Name changed again'
product.save(update_fields=['name'])
use F
Reporter.objects.update(stories_filed=F('stories_filed') + 1)
4. 缓存: local memory or redis
没有加一层缓存解决不了的问题, 如果有, 加两层……
有些数据库查询, 每次请求都会触发; 有些请求会触发 N 次相同的重复查询;
这些不一定能通过上面列举的方法优化
可以根据业务场景, 考虑是否能加缓存
细粒度:
from django.core.cache import caches
def get_default_department_id_from_cache() -> int:
cache = caches["locmem"]
id = cache.get(DEFAULT_DEPT_ID)
if id is not None:
return id
id = Department.objects.get_default().id
cache.set(DEFAULT_DEPT_ID, category_id, 1 * 60)
return id