每天40分玩转Django：Django性能优化

Django性能优化

一、性能优化要点总览表

优化类别具体措施预期结果数据库优化索引优化、查询优化、N+1问题提拔查询速率、淘汰数据库负载缓存优化Redis缓存、页面缓存、查询缓存淘汰数据库访问、提拔响应速率异步处理Celery任务队列、异步视图、异步ORM提拔并发性能、优化用户体验 二、数据库优化实现

1. 模子设计和索引优化

1. # models.py
2. from django.db import models
3. from django.db.models import Index
5. class Category(models.Model):
6.     name = models.CharField(max_length=100)
7.     slug = models.SlugField(unique=True)
8.    
9.     class Meta:
10.         indexes = [
11.             models.Index(fields=['name']),
12.             models.Index(fields=['slug'])
13.         ]
15. class Article(models.Model):
16.     title = models.CharField(max_length=200)
17.     content = models.TextField()
18.     category = models.ForeignKey(Category, on_delete=models.CASCADE)
19.     author = models.ForeignKey('auth.User', on_delete=models.CASCADE)
20.     created_at = models.DateTimeField(auto_now_add=True)
21.     updated_at = models.DateTimeField(auto_now=True)
22.     views = models.IntegerField(default=0)
23.    
24.     class Meta:
25.         indexes = [
26.             models.Index(fields=['created_at', '-views']),
27.             models.Index(fields=['category', 'author']),
28.         ]

复制代码

2. 查询优化

1. # views.py
2. from django.db.models import Prefetch, Count, Q
3. from django.views.generic import ListView
5. class ArticleListView(ListView):
6.     model = Article
7.     template_name = 'articles/article_list.html'
8.     context_object_name = 'articles'
9.    
10.     def get_queryset(self):
11.         # 使用select_related减少外键查询
12.         queryset = Article.objects.select_related(
13.             'category', 'author'
14.         ).prefetch_related(
15.             'tags',
16.             Prefetch(
17.                 'comments',
18.                 queryset=Comment.objects.select_related('user')
19.             )
20.         )
21.         
22.         # 添加过滤条件
23.         category = self.request.GET.get('category')
24.         if category:
25.             queryset = queryset.filter(category__slug=category)
26.             
27.         # 添加聚合查询
28.         queryset = queryset.annotate(
29.             comment_count=Count('comments'),
30.             like_count=Count('likes')
31.         )
32.         
33.         return queryset.order_by('-created_at')

复制代码

3. 批量操纵优化

1. # utils.py
2. from django.db import transaction
4. def bulk_create_articles(articles_data):
5.     """批量创建文章"""
6.     with transaction.atomic():
7.         articles = [
8.             Article(
9.                 title=data['title'],
10.                 content=data['content'],
11.                 category_id=data['category_id'],
12.                 author_id=data['author_id']
13.             )
14.             for data in articles_data
15.         ]
16.         return Article.objects.bulk_create(articles)
18. def bulk_update_views(article_ids):
19.     """批量更新文章浏览量"""
20.     with transaction.atomic():
21.         Article.objects.filter(id__in=article_ids).update(
22.             views=models.F('views') + 1
23.         )

复制代码

三、缓存优化实现

1. Redis缓存配置

1. # settings.py
2. CACHES = {
3.     'default': {
4.         'BACKEND': 'django_redis.cache.RedisCache',
5.         'LOCATION': 'redis://127.0.0.1:6379/1',
6.         'OPTIONS': {
7.             'CLIENT_CLASS': 'django_redis.client.DefaultClient',
8.         }
9.     }
10. }
12. # 使用Redis作为Session后端
13. SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
14. SESSION_CACHE_ALIAS = 'default'

复制代码

2. 视图缓存实现

1. # views.py
2. from django.views.decorators.cache import cache_page
3. from django.utils.decorators import method_decorator
4. from django.core.cache import cache
5. from django.conf import settings
7. @method_decorator(cache_page(60 * 15), name='dispatch')
8. class CategoryListView(ListView):
9.     model = Category
10.     template_name = 'categories/category_list.html'
12. class ArticleDetailView(DetailView):
13.     model = Article
14.    
15.     def get_object(self):
16.         article_id = self.kwargs['pk']
17.         cache_key = f'article_{article_id}'
18.         
19.         # 尝试从缓存获取
20.         article = cache.get(cache_key)
21.         if article is None:
22.             # 缓存未命中，从数据库获取
23.             article = super().get_object()
24.             # 存入缓存
25.             cache.set(cache_key, article, timeout=60*30)
26.         
27.         return article
29. def get_popular_articles():
30.     cache_key = 'popular_articles'
31.     articles = cache.get(cache_key)
32.    
33.     if articles is None:
34.         articles = Article.objects.annotate(
35.             total_score=Count('likes') + Count('comments')
36.         ).order_by('-total_score')[:10]
37.         cache.set(cache_key, articles, timeout=60*60)
38.    
39.     return articles

复制代码

3. 缓存装饰器

1. # decorators.py
2. from functools import wraps
3. from django.core.cache import cache
5. def cache_result(timeout=300):
6.     def decorator(func):
7.         @wraps(func)
8.         def wrapper(*args, **kwargs):
9.             # 生成缓存键
10.             cache_key = f"{func.__name__}:{args}:{kwargs}"
11.             result = cache.get(cache_key)
12.             
13.             if result is None:
14.                 result = func(*args, **kwargs)
15.                 cache.set(cache_key, result, timeout=timeout)
16.             
17.             return result
18.         return wrapper
19.     return decorator
21. # 使用示例
22. @cache_result(timeout=60*5)
23. def get_article_stats(article_id):
24.     return {
25.         'views': Article.objects.get(id=article_id).views,
26.         'comments': Comment.objects.filter(article_id=article_id).count(),
27.         'likes': Like.objects.filter(article_id=article_id).count()
28.     }

复制代码

四、异步处理实现

1. Celery配置和任务

1. # celery.py
2. from celery import Celery
3. import os
5. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
7. app = Celery('myproject')
8. app.config_from_object('django.conf:settings', namespace='CELERY')
9. app.autodiscover_tasks()
11. # tasks.py
12. from celery import shared_task
13. from django.core.mail import send_mail
14. from .models import Article
16. @shared_task
17. def send_article_notification(article_id):
18.     article = Article.objects.get(id=article_id)
19.     subscribers = article.category.subscribers.all()
20.    
21.     for subscriber in subscribers:
22.         send_mail(
23.             f'新文章: {article.title}',
24.             f'查看最新文章：{article.get_absolute_url()}',
25.             'noreply@example.com',
26.             [subscriber.email],
27.             fail_silently=False,
28.         )
30. @shared_task
31. def update_article_stats():
32.     """定期更新文章统计信息"""
33.     for article in Article.objects.all():
34.         stats = get_article_stats(article.id)
35.         article.stats_cache = stats
36.         article.save()

复制代码

2. 异步视图实现

1. # views.py
2. from django.http import JsonResponse
3. from asgiref.sync import sync_to_async
4. from channels.layers import get_channel_layer
5. from asgiref.sync import async_to_sync
7. async def async_article_detail(request, pk):
8.     article = await sync_to_async(Article.objects.get)(id=pk)
9.     context = {
10.         'title': article.title,
11.         'content': article.content,
12.         'author': article.author.username
13.     }
14.     return JsonResponse(context)
16. class CommentCreateView(View):
17.     def post(self, request, article_id):
18.         comment = Comment.objects.create(
19.             article_id=article_id,
20.             user=request.user,
21.             content=request.POST['content']
22.         )
23.         
24.         # 异步发送WebSocket通知
25.         channel_layer = get_channel_layer()
26.         async_to_sync(channel_layer.group_send)(
27.             f"article_{article_id}",
28.             {
29.                 "type": "comment.notification",
30.                 "message": {
31.                     "comment_id": comment.id,
32.                     "user": comment.user.username,
33.                     "content": comment.content
34.                 }
35.             }
36.         )
37.         
38.         return JsonResponse({'status': 'success'})

复制代码

3. 异步中央件

1. # middleware.py
2. from django.core.cache import cache
3. from asgiref.sync import sync_to_async
5. class AsyncCacheMiddleware:
6.     def __init__(self, get_response):
7.         self.get_response = get_response
8.    
9.     async def __call__(self, request):
10.         cache_key = f"page_cache:{request.path}"
11.         response = await sync_to_async(cache.get)(cache_key)
12.         
13.         if response is None:
14.             response = await self.get_response(request)
15.             await sync_to_async(cache.set)(
16.                 cache_key,
17.                 response,
18.                 timeout=300
19.             )
20.         
21.         return response

复制代码

五、性能监控

1. 查询日志记录

1. # middleware.py
2. import time
3. import logging
4. from django.db import connection
6. logger = logging.getLogger(__name__)
8. class QueryCountMiddleware:
9.     def __init__(self, get_response):
10.         self.get_response = get_response
11.    
12.     def __call__(self, request):
13.         start_time = time.time()
14.         initial_queries = len(connection.queries)
15.         
16.         response = self.get_response(request)
17.         
18.         total_time = time.time() - start_time
19.         total_queries = len(connection.queries) - initial_queries
20.         
21.         if total_queries > 10:  # 设置查询数量阈值
22.             logger.warning(
23.                 f'Path: {request.path} - '
24.                 f'Queries: {total_queries} - '
25.                 f'Time: {total_time:.2f}s'
26.             )
27.         
28.         return response

复制代码

六、性能优化流程图

最佳实践建议：

• 数据库优化：
  
  
  
  • 公道使用索引
    
  • 避免N+1查询问题
    
  • 使用批量操纵替换循环操纵
    
  • 定期分析和优化慢查询
    
  
  
• 缓存策略：
  
  
  
  • 公道设置缓存时间
    
  • 分层缓存架构
    
  • 实时更新缓存
    
  • 避免缓存雪崩
    
  
  
• 异步处理：
  
  
  
  • 公道使用任务队列
    
  • 异步处理耗时操纵
    
  • 恰当使用异步视图
    
  • 监控任务实验状态
    
  
  

这就是关于Django性能优化的详细内容。通过实践这些优化策略，你可以明显提拔Django应用的性能。如果有任何问题，接待随时提出！

---

怎么样今天的内容还满意吗？再次感谢朋友们的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财政自由，还请给个赞，谢谢！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。