iOS用户拒通知权限后，如何用Django Channels实现Django+React Native定向通知？

场景适配性 #

这个场景完全适合使用Django Channels。当iOS用户拒绝推送权限后，应用内实时通知是最优替代方案，而Django Channels基于WebSocket实现的长连接，正好满足这种需要实时、定向推送消息的需求，能确保用户打开应用时即时收到通知。

核心运作机制说明 #

Django Channel Layers是跨进程/实例的消息路由系统，允许不同WebSocket Consumer实例间通信。你可以将每个用户的WebSocket连接加入专属分组（比如以用户ID命名的组），当需要给目标用户发通知时，直接向该分组发送消息，所有加入该组的用户设备连接都会同步收到消息。

具体实现步骤（结合你的代码修改） #

1. 配置Channel Layers #

首先在项目settings.py中配置Channel Layers，生产环境建议用Redis，开发阶段可临时用内存版：

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
    # 开发测试用（生产禁用）：
    # "BACKEND": "channels.layers.InMemoryChannelLayer"
}

2. 修改React Native客户端代码 #

• 连接时传递用户身份（示例用URL参数，生产建议用Token验证）
• 添加消息监听逻辑，接收并展示应用内通知

// 假设当前登录用户ID为currentUserId
const ws = new WebSocket(`ws://your-domain/notif-socket/?user_id=${currentUserId}`);

useEffect(() => {
    ws.onopen = () => {
      console.log('Connected!');
    };

    ws.onerror = e => {
      console.log(e.message);
    };

    // 监听服务端推送的通知
    ws.onmessage = (event) => {
      const notification = JSON.parse(event.data);
      // 这里替换为你的应用内通知展示逻辑（比如自定义弹窗、通知中心）
      Alert.alert(notification.title, notification.body);
    };

    // 组件卸载时关闭连接
    return () => {
      ws.close();
    };
}, []);

// 发送通知的代码补充from_user字段
const data = {
   from_user: currentUserId.toString(),
   to_user: user_id.toString(),
   title: 'Notification',
   body: 'User_2 liked your post',
   image: imageURL,
};

ws.send(JSON.stringify(data));

3. 修改Django Consumer代码 #

• 连接时验证用户身份，将用户加入专属分组
• 接收消息后保存到数据库，并通过Channel Layers推送给目标用户

# consumers.py
import json
from channels.generic.websocket import WebsocketConsumer
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
from django.contrib.auth.models import User
from .models import Notification

class NotificationConsumer(WebsocketConsumer):

    def connect(self):
        # 从URL参数获取当前连接用户ID
        self.user_id = self.scope['query_params'].get('user_id')
        if not self.user_id:
            self.close()
            return
        
        # 加入用户专属分组
        self.group_name = f'user_{self.user_id}'
        async_to_sync(self.channel_layer.group_add)(
            self.group_name,
            self.channel_name
        )
        self.accept()

    def disconnect(self, close_code):
        # 断开连接时移除分组
        async_to_sync(self.channel_layer.group_discard)(
            self.group_name,
            self.channel_name
        )

    def receive(self, text_data):
        data = json.loads(text_data)
        # 保存通知到数据库
        try:
            from_user = User.objects.get(id=data['from_user'])
            to_user = User.objects.get(id=data['to_user'])
            Notification.objects.create(
                from_user=from_user,
                to_user=to_user,
                title=data['title'],
                body=data['body'],
                image=data.get('image')
            )
        except User.DoesNotExist:
            self.send(text_data=json.dumps({'error': 'User not found'}))
            return
        
        # 推送通知到目标用户的分组
        async_to_sync(self.channel_layer.group_send)(
            f'user_{data["to_user"]}',
            {
                'type': 'send_notification',
                'data': data
            }
        )

    # 定义通知发送处理方法
    def send_notification(self, event):
        data = event['data']
        self.send(text_data=json.dumps({
            'title': data['title'],
            'body': data['body'],
            'image': data.get('image')
        }))

4. 配置WebSocket路由 #

在项目routing.py中添加WebSocket路由规则：

# routing.py
from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'notif-socket/$', consumers.NotificationConsumer.as_asgi()),
]

额外注意事项 #

• 身份验证：生产环境不要直接用URL传递用户ID，建议用JWT Token，通过请求头或连接后发送的消息传递Token，服务端验证后获取用户ID。
• 离线处理：如果目标用户未在线，可在数据库标记通知为未读，等用户下次连接时主动推送所有未读通知。
• 生产环境要求：必须使用Redis作为Channel Layer后端，内存版仅适用于开发测试，多进程/实例部署时无法互通消息。

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