Django 的数据库操作（ORM）虽然方便，但如果使用不当，很容易出现性能问题（比如查询缓慢、数据库压力大）。数据库优化的核心目标是：减少不必要的查询、减少数据传输量、让查询跑得更快。
1、N+1查询问题
当查询包含外键关联的数据时，如果循环获取关联对象，会产生 “1 次主查询 + N 次关联查询” 的低效操作（N 是主查询结果的数量）。- # models.py
- class Author(models.Model):
- name = models.CharField(max_length=100)
- age = models.IntegerField()
- class Book(models.Model):
- title = models.CharField(max_length=200)
- author = models.ForeignKey(Author, on_delete=models.CASCADE) # 外键关联作者
- # 1. 先查询所有书籍（1次查询）
- books = Book.objects.all() # SQL: SELECT * FROM book;
- # 2. 循环获取每本书的作者（N次查询，N=书籍数量）
- for book in books:
- print(book.author.name) # 每次都会触发：SELECT * FROM author WHERE id=?;
用select_related预加载关联对象，select_related适用于外键、一对一关联，会通过JOIN语句一次性把关联数据查出来。- # 1次查询搞定所有数据（主表+关联表JOIN）
- books = Book.objects.select_related('author').all() # SQL: SELECT * FROM book JOIN author ON ...;
- # 循环获取作者时，不会再查数据库
- for book in books:
- print(book.author.name) # 直接从已加载的数据中获取
prefetch_related适用于多对多 、反向外键(反向查询)关系。它先执行一次主查询，再执行一次查询来获取所有关联对象，然后在 Python 层进行"连接"，效率更高。- class Category(models.Model):
- name = models.CharField(max_length=50)
- class Book(models.Model):
- # ... 其他字段
- categories = models.ManyToManyField(Category) # 多对多关联
- # 用prefetch_related预加载多对多数据
- books = Book.objects.prefetch_related('categories').all() # SELECT ... FROM table1; SELECT ... FROM table2 WHERE id IN (...)
- categories = Category.objects.prefetch_related('book_set').all()
- for book in books:
- # 不会触发额外查询
- print([c.name for c in book.categories.all()])
默认情况下，Book.objects.all()会查询表中所有字段，但很多时候我们只需要其中几个字段
2.1 使用 only() 和 defer()
- # 只获取需要的字段
- books = Book.objects.only('title', 'publication_date') # 只获取标题和出版日期
- # 排除大字段
- books = Book.objects.defer('description') # 不获取描述字段（假设是很大的文本字段）
- # 获取字典列表
- book_titles = Book.objects.values('title', 'author__name') # 返回 [{'title': '...', 'author__name': '...'}]
- # 获取元组列表
- book_titles = Book.objects.values_list('title', flat=True) # 返回 ['标题1', '标题2', ...]
以下情况需要加索引
- 频繁用filter()、exclude()过滤的字段（比如where author_id=1）
- 频繁用order_by()排序的字段（比如order by publish_time）
- 外键字段（Django 会自动加索引）、唯一约束字段（自动加索引）
- # models.py
- class Book(models.Model):
- title = models.CharField(max_length=200, db_index=True) # 为标题添加索引
- author = models.ForeignKey(Author, on_delete=models.CASCADE)
- publication_date = models.DateField()
-
- # 复合索引
- class Meta:
- indexes = [
- models.Index(fields=['author', 'publication_date']), # 复合索引
- ]
- class Customer(models.Model):
- first_name = models.CharField(max_length=100, db_index=True)
- last_name = models.CharField(max_length=100, db_index=True)
- email = models.EmailField(unique=True) # 唯一约束自动创建索引
-
- class Meta:
- # 复合索引
- index_together = [
- ['first_name', 'last_name'],
- ]
用bulk_create（批量创建）和bulk_update（批量更新）
4.1 使用 bulk_create() 一次性创建多个对象
- # 不好的做法
- books = []
- for i in range(1000):
- book = Book(title=f"Book {i}", author=some_author)
- book.save() # 每次保存都执行一次INSERT
- # 好的做法
- books = [Book(title=f"Book {i}", author=some_author) for i in range(1000)]
- Book.objects.bulk_create(books) # 一次性执行批量INSERT
- # 批量更新
- books = Book.objects.filter(publication_date__year=2020)
- for book in books:
- book.price = book.price * 0.9 # 打9折
- Book.objects.bulk_update(books, ['price']) # 一次性批量更新
- # 使用F表达式
- Book.objects.filter(publication_date__year=2020).update(price=F('price') * 0.8)
在高并发场景下，为每个请求创建和销毁数据库连接开销很大。使用数据库连接池可以复用连接，显著提升性能。
安装django-db-connection-pool- pip install django-db-connection-pool
- DATABASES = {
- 'default': {
- 'ENGINE': 'dj_db_conn_pool.backends.mysql',
- 'NAME': 'your_db',
- 'USER': 'your_user',
- 'PASSWORD': 'your_password',
- 'HOST': 'localhost',
- 'PORT': '3306',
- 'OPTIONS': {
- 'POOL_SIZE': 20, # 连接池大小
- 'MAX_OVERFLOW': 10, # 允许超过POOL_SIZE的最大连接数
- 'POOL_RECYCLE': 3600, # 连接回收时间（秒）
- }
- }
- }
annotate和 aggregate是 Django ORM 中用于执行数据库聚合操作的两个强大工具，它们允许你在数据库层面进行计算，避免将大量数据拉到 Python 中进行处理，从而显著提升性能。
模型示例：- from django.db import models
- class Author(models.Model):
- name = models.CharField(max_length=100)
- country = models.CharField(max_length=50)
- class Book(models.Model):
- title = models.CharField(max_length=200)
- author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name='books')
- price = models.DecimalField(max_digits=6, decimal_places=2)
- published_date = models.DateField()
- rating = models.IntegerField(choices=[(i, i) for i in range(1, 6)])
annotate()为查询集中的每个对象添加计算字段（注解）。
6.1.1 基本用法：为作者添加书籍计数
- from django.db.models import Count
- # 获取所有作者，并为每个作者添加书籍数量字段
- authors = Author.objects.annotate(book_count=Count('books'))
- for author in authors:
- print(f"{author.name} 写了 {author.book_count} 本书")
- from django.db.models import Avg, Max
- # 为每个作者添加平均价格和最高评分字段
- authors = Author.objects.annotate(
- avg_price=Avg('books__price'),
- max_rating=Max('books__rating')
- )
- for author in authors:
- print(f"{author.name}: 平均价格 ${author.avg_price:.2f}, 最高评分 {author.max_rating}")
- from django.db.models import Count
- # 为每个作者添加2020年出版的书籍数量
- authors = Author.objects.annotate(
- books_2020=Count('books', filter=models.Q(books__published_date__year=2020))
- )
- for author in authors:
- print(f"{author.name} 在2020年出版了 {author.books_2020} 本书")
- from django.db.models import F, Count, Value
- # 计算每个作者的平均评分和总价值
- authors = Author.objects.annotate(
- book_count=Count('books'),
- total_value=Sum('books__price'),
- ).annotate(
- avg_rating=Avg('books__rating'),
- value_per_book=F('total_value') / F('book_count')
- )
- for author in authors:
- print(f"{author.name}: 每本书平均价值 ${author.value_per_book:.2f}")
aggregate()计算整个查询集的统计值，返回一个字典。
6.2.1 基本聚合：计算所有书籍的总价和平均价
- from django.db.models import Sum, Avg
- # 计算所有书籍的总价和平均价
- stats = Book.objects.aggregate(
- total_price=Sum('price'),
- average_price=Avg('price')
- )
- print(f"所有书籍总价: ${stats['total_price']}")
- print(f"平均价格: ${stats['average_price']:.2f}")
- from django.db.models import Max, Min
- # 获取最高和最低评分
- rating_stats = Book.objects.aggregate(
- highest_rating=Max('rating'),
- lowest_rating=Min('rating')
- )
- print(f"最高评分: {rating_stats['highest_rating']}")
- print(f"最低评分: {rating_stats['lowest_rating']}")
- from django.db.models import Count, Avg
- # 统计某位作者的书籍
- author_stats = Book.objects.filter(
- author__name="J.K. Rowling"
- ).aggregate(
- book_count=Count('id'),
- avg_rating=Avg('rating')
- )
- print(f"J.K. Rowling 写了 {author_stats['book_count']} 本书")
- print(f"平均评分: {author_stats['avg_rating']:.1f}")
7.1 用count()和exists()代替全量查询
- # 好：直接查数量（高效）
- total = Book.objects.count()
- # 差：先查所有数据再算长度（低效，尤其数据量大时）
- total = len(Book.objects.all())
- # 好：存在即返回True（查到1条就停止）
- has_book = Book.objects.filter(title='Django入门').exists()
- # 差：查所有数据再判断（可能查很多条）
- has_book = len(Book.objects.filter(title='Django入门')) > 0
如果 ORM 查询太复杂（比如多表复杂 JOIN），可以用原生 SQL：- from django.db import connection
- def get_book_stats():
- with connection.cursor() as cursor:
- cursor.execute("""
- SELECT author.name, COUNT(book.id)
- FROM author
- JOIN book ON author.id = book.author_id
- GROUP BY author.id
- """)
- # 获取查询结果（(作者名, 书籍数量), ...）
- result = cursor.fetchall()
- return result
|
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
照妖镜
