Django Channels：Token过期或失效时服务端主动断开WebSocket连接

一、定期校验Token与数据变更，主动断开无效连接 #

WebSocket是长连接，建立时的JWT认证只做了一次校验，后续Token过期或用户数据（比如所属公司）变更后，旧连接仍会留在原频道。可以在Consumer里加定时校验逻辑，主动清理无效连接：

import asyncio
import json
from django.contrib.auth.models import User
from rest_framework_simplejwt.tokens import AccessToken
from channels.generic.websocket import AsyncWebsocketConsumer

class CompanyConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # 从查询参数取JWT并完成初始认证
        query_params = self.scope['query_string'].decode().split('&')
        self.token = next(p.split('=')[1] for p in query_params if p.startswith('token='))
        try:
            token = AccessToken(self.token)
            self.user = await User.objects.aget(id=token['user_id'])
            self.company_name = self.user.profile.company.name
            # 加入对应公司频道组
            await self.channel_layer.group_add(
                self.company_name,
                self.channel_name
            )
            await self.accept()
            # 启动定时校验任务（每5分钟一次）
            asyncio.create_task(self.validate_connection_loop())
        except Exception:
            await self.close(code=4001)  # 初始认证失败直接关闭

    async def validate_connection_loop(self):
        while True:
            await asyncio.sleep(300)
            try:
                # 重新校验Token有效性（AccessToken解析时自动校验过期）
                AccessToken(self.token).verify()
                # 校验用户当前所属公司是否和连接时一致
                current_company = await self.user.profile.aget().company.name
                if current_company != self.company_name:
                    # 数据变更，主动断开连接
                    await self.disconnect(code=4011)
                    break
            except Exception:
                # Token过期或无效，断开连接
                await self.disconnect(code=4010)
                break

    async def disconnect(self, code):
        # 断开前移除频道组
        await self.channel_layer.group_discard(self.company_name, self.channel_name)
        await super().disconnect(code)

二、数据变更时实时清理连接 #

当管理员修改用户所属公司这类关键数据时，直接触发清理该用户的所有WebSocket连接，不用等定时校验：

1. 连接建立时记录用户与通道的关联 #

在Consumer的connect方法里，把用户ID和channel_name存在Redis（Channels默认用Redis做通道层）：

await self.channel_layer.redis_client.hset(
    f"user_channels:{self.user.id}",
    self.channel_name,
    self.company_name
)

2. 用模型信号触发连接清理 #

在用户关联数据（比如UserProfile）修改时，通过信号主动清理连接：

from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import UserProfile
from channels.layers import get_channel_layer
import asyncio

@receiver(post_save, sender=UserProfile)
def handle_company_change(sender, instance, created, **kwargs):
    if not created:
        # 对比修改前后的公司ID（需要在模型里存旧值）
        old_company_id = instance.__original_company_id
        new_company_id = instance.company_id
        if old_company_id != new_company_id:
            channel_layer = get_channel_layer()
            user_id = instance.user.id
            # 获取该用户的所有活跃连接
            user_channels = channel_layer.redis_client.hgetall(f"user_channels:{user_id}")
            
            for channel_name, old_company in user_channels.items():
                channel_name = channel_name.decode()
                old_company = old_company.decode()
                # 从旧公司频道组移除
                asyncio.run(channel_layer.group_discard(old_company, channel_name))
                # 主动发送断开指令
                asyncio.run(channel_layer.send(channel_name, {
                    "type": "disconnect",
                    "code": 4011
                }))
            # 清理Redis里的用户连接记录
            channel_layer.redis_client.delete(f"user_channels:{user_id}")

3. 模型里保存旧值用于对比 #

在UserProfile模型里重写save方法，记录修改前的公司ID：

class UserProfile(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)
    company = models.ForeignKey(Company, on_delete=models.SET_NULL, null=True)

    def save(self, *args, **kwargs):
        if self.pk:
            # 保存修改前的公司ID
            self.__original_company_id = UserProfile.objects.get(pk=self.pk).company_id
        super().save(*args, **kwargs)

三、推送前校验权限，避免发错数据 #

如果不想直接断开连接，也可以在推送消息前做最后校验，确保用户有权限接收：

async def send_company_notification(self, event):
    message = event['message']
    # 推送前再次校验用户当前所属公司
    current_company = await self.user.profile.aget().company.name
    if current_company != self.company_name:
        return  # 权限不符，不发送消息
    await self.send(text_data=json.dumps(message))

关于Token过期的风险补充 #

你提到的过期Token无法建立新连接的风险确实很低，但要注意：

• 把AccessToken的有效期设短一些（比如15-30分钟），配合定时校验间隔；
• 不要在WebSocket里用Refresh Token，只保留AccessToken用于权限校验；
• 定时校验的间隔要小于Token有效期，避免出现长时间的无效连接。

内容的提问来源于stack exchange，提问作者Stefan Colic