Django从零开始（八）：中间件、信号与缓存

Django 5 小时前 0 2

一、中间件（Middleware）

中间件是Django请求/响应处理的"洋葱"式管道。每个中间件可以在请求到达视图之前或响应返回给客户端之后执行操作。

请求处理流程

# Django中间件执行顺序（洋葱模型）
# 请求方向（从外到内）：
Request → Middleware1.process_request → Middleware2.process_request → View

# 响应方向（从内到外）：
View → Middleware2.process_response → Middleware1.process_response → Response

1.1 基于函数的中间件

# blog/middleware.py

import time
import logging
from django.http import JsonResponse

logger = logging.getLogger(__name__)


def simple_middleware(get_response):
    """简单中间件模板"""

    # 一次性设置和初始化（服务器启动时执行一次）
    print("中间件初始化")

    def middleware(request):
        # 视图执行前的代码
        start_time = time.time()

        # ====== 请求预处理 ======

        # 记录请求信息
        logger.info(f"[{request.method}] {request.path} from {request.META.get('REMOTE_ADDR')}")

        response = get_response(request)  # 调用下一个中间件或视图

        # ====== 响应后处理 ======

        # 记录响应时间
        duration = time.time() - start_time
        logger.info(f"Request to {request.path} took {duration:.3f}s")

        return response

    return middleware


def maintenance_mode_middleware(get_response):
    """维护模式中间件"""
    def middleware(request):
        from django.conf import settings

        # 检查是否处于维护模式
        if getattr(settings, 'MAINTENANCE_MODE', False):
            # 管理员仍然可以访问
            if request.user.is_authenticated and request.user.is_superuser:
                return get_response(request)

            # API请求返回JSON
            if request.path.startswith('/api/'):
                return JsonResponse(
                    {'error': '系统维护中，请稍后访问'},
                    status=503
                )

            # 普通请求返回维护页面
            from django.shortcuts import render
            return render(request, 'maintenance.html', status=503)

        return get_response(request)
    return middleware


def ip_blacklist_middleware(get_response):
    """IP黑名单中间件"""
    def middleware(request):
        from django.conf import settings

        blacklisted_ips = getattr(settings, 'IP_BLACKLIST', [])
        client_ip = request.META.get('REMOTE_ADDR')
        forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')

        # 获取真实IP（处理反向代理）
        if forwarded_for:
            client_ip = forwarded_for.split(',')[0].strip()

        if client_ip in blacklisted_ips:
            logger.warning(f"Blocked request from blacklisted IP: {client_ip}")
            return JsonResponse({'error': 'Access denied'}, status=403)

        return get_response(request)
    return middleware

1.2 基于类的中间件

class RequestTimingMiddleware:
    """请求计时中间件（类版本）"""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 请求前处理
        start_time = time.time()
        request.start_time = start_time

        response = self.get_response(request)

        # 响应后处理
        duration = time.time() - start_time
        response['X-Request-Duration'] = f'{duration:.3f}s'

        # 慢请求警告
        if duration > 1.0:
            logger.warning(
                f"Slow request: {request.method} {request.path} took {duration:.3f}s"
            )

        return response

    def process_exception(self, request, exception):
        """异常处理"""
        logger.error(
            f"Exception in {request.method} {request.path}: {str(exception)}",
            exc_info=True
        )
        return None  # 返回None让Django继续正常异常处理

    def process_view(self, request, view_func, view_args, view_kwargs):
        """视图执行前调用"""
        # 可以在这里做权限检查等
        pass


class CORSMiddleware:
    """跨域请求中间件"""
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)

        # 设置CORS头
        response['Access-Control-Allow-Origin'] = '*'
        response['Access-Control-Allow-Methods'] = 'GET, POST, PUT, PATCH, DELETE, OPTIONS'
        response['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
        response['Access-Control-Max-Age'] = '86400'

        return response


class ActiveUserMiddleware:
    """记录在线用户中间件"""
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        if request.user.is_authenticated:
            from django.utils import timezone
            from django.core.cache import cache

            # 使用缓存记录活跃用户（5分钟过期）
            cache_key = f'active_user_{request.user.id}'
            cache.set(cache_key, timezone.now().isoformat(), timeout=300)

        return self.get_response(request)

1.3 注册中间件

# settings.py

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',

    # 自定义中间件（顺序很重要！）
    'blog.middleware.RequestTimingMiddleware',
    'blog.middleware.CORSMiddleware',
    'blog.middleware.maintenance_mode_middleware',
    'blog.middleware.ActiveUserMiddleware',
]

二、信号（Signals）

信号允许某些发送者通知一组接收者某个动作已经发生。它是一种松耦合的事件处理机制。

2.1 Django内置信号

# blog/signals.py

from django.db.models.signals import (
    pre_save, post_save,
    pre_delete, post_delete,
    m2m_changed,
    request_started, request_finished,
)
from django.dispatch import receiver
from django.contrib.auth.models import User
from django.utils import timezone
from .models import Article, Comment, Profile


# ===== 模型信号 =====

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    """用户创建后自动创建Profile"""
    if created:
        Profile.objects.create(user=instance)


@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
    """用户保存时保存Profile"""
    if hasattr(instance, 'profile'):
        instance.profile.save()


@receiver(pre_save, sender=Article)
def article_pre_save(sender, instance, **kwargs):
    """文章保存前自动处理"""
    # 自动生成slug
    if not instance.slug:
        from django.utils.text import slugify
        base_slug = slugify(instance.title)
        slug = base_slug
        counter = 1
        while Article.objects.filter(slug=slug).exclude(pk=instance.pk).exists():
            slug = f'{base_slug}-{counter}'
            counter += 1
        instance.slug = slug

    # 发布时间
    if instance.status == 'published' and not instance.published_at:
        instance.published_at = timezone.now()


@receiver(post_save, sender=Article)
def article_post_save(sender, instance, created, **kwargs):
    """文章保存后的操作"""
    if created:
        # 新文章通知
        import logging
        logger = logging.getLogger(__name__)
        logger.info(f"新文章创建: {instance.title} by {instance.author.username}")
    else:
        # 更新搜索索引等
        pass


@receiver(post_delete, sender=Article)
def article_post_delete(sender, instance, **kwargs):
    """文章删除后清理"""
    import logging
    logger = logging.getLogger(__name__)
    logger.info(f"文章已删除: {instance.title}")


@receiver(m2m_changed, sender=Article.tags.through)
def article_tags_changed(sender, instance, action, **kwargs):
    """文章标签变更"""
    if action == 'post_add':
        # 标签添加后的操作
        pass


# ===== 请求信号 =====

@receiver(request_started)
def on_request_started(sender, **kwargs):
    """请求开始"""
    pass

@receiver(request_finished)
def on_request_finished(sender, **kwargs):
    """请求结束"""
    pass

2.2 自定义信号

# blog/signals.py

import django.dispatch

# 定义自定义信号
article_published = django.dispatch.Signal()  # 文章发布信号
comment_approved = django.dispatch.Signal()   # 评论审核通过信号


# ===== 发送信号 =====

class ArticleViewSet(viewsets.ModelViewSet):
    @action(detail=True, methods=['post'])
    def publish(self, request, pk=None):
        article = self.get_object()
        article.status = 'published'
        article.published_at = timezone.now()
        article.save()

        # 发送文章发布信号
        article_published.send(
            sender=article.__class__,
            article=article,
            publisher=request.user,
        )
        return Response({'status': 'published'})


# ===== 接收信号 =====

@receiver(article_published)
def on_article_published(sender, article, publisher, **kwargs):
    """文章发布后的操作"""
    # 1. 发送通知给订阅者
    notify_subscribers(article)

    # 2. 更新搜索索引
    update_search_index(article)

    # 3. 清除缓存
    clear_article_cache(article)

    # 4. 记录日志
    logger.info(f"文章已发布: {article.title} by {publisher.username}")


@receiver(article_published)
def send_publish_notification(sender, article, publisher, **kwargs):
    """发送发布通知"""
    from django.core.mail import mail_admins
    mail_admins(
        subject=f'新文章发布: {article.title}',
        message=f'{publisher.username} 发布了新文章 "{article.title}"',
    )

2.3 注册信号

# blog/apps.py

from django.apps import AppConfig

class BlogConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'blog'

    def ready(self):
        """应用启动时导入信号"""
        import blog.signals

三、缓存（Caching）

3.1 缓存配置

# settings.py

# 方式1：内存缓存（开发环境）
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        'LOCATION': 'unique-snowflake',
    }
}

# 方式2：Redis缓存（生产环境推荐）
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django.core.cache.backends.redis.RedisCache',
        },
        'KEY_PREFIX': 'blog_',
        'TIMEOUT': 300,  # 默认缓存5分钟
    }
}

# 方式3：文件系统缓存
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
        'LOCATION': '/var/tmp/django_cache',
    }
}

# 缓存超时设置
CACHE_TTL = 60 * 5  # 5分钟

3.2 视图缓存

from django.views.decorators.cache import cache_page, cache_control, never_cache
from django.views.decorators.vary import vary_on_headers, vary_on_cookie


# 缓存整个视图（15分钟）
@cache_page(60 * 15)
def article_list(request):
    articles = Article.objects.filter(status='published')
    return render(request, 'blog/article_list.html', {'articles': articles})


# 缓存控制
@cache_control(max_age=3600, public=True)
def public_page(request):
    return render(request, 'public.html')


# 禁用缓存
@never_cache
def user_dashboard(request):
    return render(request, 'dashboard.html')


# 根据请求头变化缓存
@cache_page(60 * 15)
@vary_on_headers('User-Agent')
def api_data(request):
    return JsonResponse({'data': '...'})


# 类视图缓存
from django.utils.decorators import method_decorator

@method_decorator(cache_page(60 * 15), name='dispatch')
class ArticleListView(ListView):
    model = Article
    template_name = 'blog/article_list.html'

3.3 手动缓存操作

from django.core.cache import cache


# 设置缓存
cache.set('article_list_key', articles, timeout=300)  # 5分钟过期
cache.set('user_stats', {'views': 1000, 'likes': 50})

# 获取缓存
articles = cache.get('article_list_key')
articles = cache.get_or_set('key', default_value, timeout=300)

# 删除缓存
cache.delete('article_list_key')
cache.delete_many(['key1', 'key2', 'key3'])

# 清空所有缓存
cache.clear()

# 检查缓存是否存在
if cache.has_key('article_list_key'):
    articles = cache.get('article_list_key')

# 批量操作
cache.set_many({'key1': 'val1', 'key2': 'val2'}, timeout=300)
result = cache.get_many(['key1', 'key2'])  # {'key1': 'val1', 'key2': 'val2'}


# 实际使用示例
def get_article_detail(request, slug):
    cache_key = f'article:{slug}'

    article = cache.get(cache_key)
    if article is None:
        article = get_object_or_404(Article, slug=slug, status='published')
        # 缓存30分钟
        cache.set(cache_key, article, timeout=60 * 30)

    return render(request, 'blog/article_detail.html', {'article': article})

3.4 模板片段缓存

{% load cache %}

{# 缓存侧边栏500秒 #}
{% cache 500 sidebar %}
    <div class="sidebar">
        <h3>热门文章</h3>
        {% for article in hot_articles %}
        <a href="{{ article.get_absolute_url }}">{{ article.title }}</a>
        {% endfor %}
    </div>
{% endcache %}

{# 带变量名的缓存（按用户区分） #}
{% cache 500 user_menu request.user.username %}
    <div class="user-menu">
        <p>{{ request.user.username }}</p>
    </div>
{% endcache %}

四、总结

本章我们学习了Django三大高级机制：

中间件：请求/响应处理管道，用于计时、CORS、维护模式、IP黑名单等
信号：松耦合的事件处理，模型CRUD触发、自定义信号
缓存：Redis缓存、视图缓存、手动缓存、模板片段缓存

下一章我们将学习 Django测试、调试与安全防护，确保你的应用稳定可靠。

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

Django教程