如何优化Django中用户获取房间消息时标记未读消息为已读的代码？

需求说明 #

当用户获取房间消息列表时，需将所有未读消息的seen字段设为True，同时通过Seen表记录每个用户对消息的查看记录，现寻求当前实现代码的改进与优化方案。

表结构与关联关系 #

共包含user、room、message、seen四张表，关联关系如下：

• 一个用户可拥有多个房间，一个房间可包含多个用户（多对多关联）；
• 一个用户可发送多条消息，一条消息仅归属一个特定用户（一对多关联）；
• 一条消息可被多个用户查看，查看记录通过seen表存储（多对多关联的中间表）。

现有模型代码（models.py） #

class User(models.Model):
    username = models.CharField(max_length=255, unique=True)
    is_active = models.BooleanField(default=False)
    date_join = models.DateTimeField(auto_now_add=True)
    is_admin = models.BooleanField(default=False)

    def __str__(self):
        return f"{self.username}|{self.is_admin}"

class Room(models.Model):
    name = models.CharField(max_length=255)
    users = models.ManyToManyField(User, related_name="rooms")

    def __str__(self):
        return f"{self.name}"

class Message(models.Model):
    content = models.TextField()
    seen = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="messages")
    room = models.ForeignKey(Room, on_delete=models.CASCADE, related_name="room_messages")
    replied_to = models.ForeignKey('self', null=True, blank=True, on_delete=models.SET_NULL)

    def __str__(self):
        return f"{self.user.id}: {self.content} [{self.created_at}]"

class Seen(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    message = models.ForeignKey(Message, on_delete=models.CASCADE, related_name="seens")
    date_seen = models.DateTimeField(auto_now_add=True)

现有视图代码（views.py） #

class MessageViewSet(GenericViewSet, ListModelMixin, RetrieveModelMixin, DestroyModelMixin):

    def list(self, request, *args, **kwargs):

        queryset = self.get_queryset()
        if self.kwargs.get("room_pk"):
            queryset = queryset.filter(room_id=self.kwargs.get("room_pk"))

        filter_queryset = self.filter_queryset(queryset)
        page = self.paginate_queryset(filter_queryset)

        if page is not None:
            serializer = self.get_serializer(page, many=True)
            response = self.get_paginated_response(serializer.data)
            for message in page:
                if message.user.id != self.request.user.id:
                    if not message.seens.select_related("message").select_related("user").filter(user_id=self.request.user.id).exists():

                        message.seen = True
                        seen_obj = Seen()
                        seen_obj.user = self.request.user
                        seen_obj.message = message
                        seen_obj.save()
                        message.seens.add(seen_obj)
                        message.save()
                else:
                    continue
            return response

代码优化与改进方案 #

1. 批量操作替代循环单条处理 #

当前代码存在N+1数据库操作问题，循环处理每条消息会产生大量查询和写入请求。改用批量操作可大幅提升性能：

def list(self, request, *args, **kwargs):
    queryset = self.get_queryset()
    room_pk = self.kwargs.get("room_pk")
    if room_pk:
        queryset = queryset.filter(room_id=room_pk)

    filter_queryset = self.filter_queryset(queryset)
    page = self.paginate_queryset(filter_queryset)

    if page is not None:
        serializer = self.get_serializer(page, many=True)
        response = self.get_paginated_response(serializer.data)

        # 提取当前页面中不属于当前用户的消息ID
        message_ids = [msg.id for msg in page if msg.user.id != request.user.id]
        if message_ids:
            # 查询当前用户已标记为已读的消息ID
            seen_message_ids = Seen.objects.filter(
                user=request.user,
                message_id__in=message_ids
            ).values_list('message_id', flat=True)
            # 筛选未读消息ID
            unseen_message_ids = [mid for mid in message_ids if mid not in seen_message_ids]

            if unseen_message_ids:
                # 批量创建Seen记录
                Seen.objects.bulk_create([
                    Seen(user=request.user, message_id=mid)
                    for mid in unseen_message_ids
                ])
                # 批量更新Message的seen字段
                Message.objects.filter(id__in=unseen_message_ids).update(seen=True)

        return response

2. 移除冗余操作与无效查询 #

• 原代码中message.seens.add(seen_obj)属于冗余操作，创建Seen对象时已通过外键关联消息，无需额外调用add；
• message.seens.select_related("message")是无效查询，seens本身就是Message的反向关联，无需再关联自身。

3. 数据库索引优化 #

为Seen表添加联合唯一索引，防止重复创建已读记录并加快查询速度：

class Seen(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    message = models.ForeignKey(Message, on_delete=models.CASCADE, related_name="seens")
    date_seen = models.DateTimeField(auto_now_add=True)

    class Meta:
        unique_together = ('user', 'message')  # 避免同一用户重复标记同一条消息
        indexes = [
            models.Index(fields=['user', 'message']),
        ]

4. 分离已读标记逻辑 #

将已读标记逻辑封装为独立方法，提升代码复用性与可读性：

def mark_messages_as_seen(self, user, message_ids):
    seen_message_ids = Seen.objects.filter(user=user, message_id__in=message_ids).values_list('message_id', flat=True)
    unseen_ids = [mid for mid in message_ids if mid not in seen_message_ids]
    if unseen_ids:
        Seen.objects.bulk_create([Seen(user=user, message_id=mid) for mid in unseen_ids])
        Message.objects.filter(id__in=unseen_ids).update(seen=True)

# 在list方法中调用
self.mark_messages_as_seen(request.user, message_ids)

5. 修正seen字段的设计缺陷 #

当前Message表的seen字段是全局状态，只要有一个用户标记已读，所有用户都会看到已读状态，不符合“多用户独立标记已读”的业务需求。建议：

• 删除Message模型中的seen字段；
• 通过Seen表单独记录每个用户的已读状态，在序列化器中动态返回当前用户的已读状态：

调整后的Message模型：

class Message(models.Model):
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="messages")
    room = models.ForeignKey(Room, on_delete=models.CASCADE, related_name="room_messages")
    replied_to = models.ForeignKey('self', null=True, blank=True, on_delete=models.SET_NULL)

    def __str__(self):
        return f"{self.user.id}: {self.content} [{self.created_at}]"

序列化器添加is_seen字段：

class MessageSerializer(serializers.ModelSerializer):
    is_seen = serializers.SerializerMethodField()

    class Meta:
        model = Message
        fields = ['id', 'content', 'created_at', 'user', 'room', 'replied_to', 'is_seen']

    def get_is_seen(self, obj):
        user = self.context['request'].user
        return obj.seens.filter(user=user).exists()

内容的提问来源于stack exchange，提问作者ARYAN-NIKNEZHAD