如何实现从Django向React Native发送定时事件？

1. 后端定时任务核心实现 #

Django 本身不适合长期运行定时任务，推荐用 Celery + Redis 搭建任务队列，结合数据库存储动态定时时间。

1.1 依赖安装与配置 #

• 安装必要包：

  pip install celery redis django-celery-beat
  

• 在 Django settings.py 中配置 Celery：

  CELERY_BROKER_URL = 'redis://localhost:6379/0'
  CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  CELERY_TIMEZONE = 'Asia/Shanghai'
  
  INSTALLED_APPS += ['django_celery_beat']
  

• 执行数据库迁移，生成定时任务相关表：

  python manage.py migrate
  

1.2 动态定时任务调度 #

假设数据库表 ScheduledEvent 存储定时事件（字段：send_time、payload、is_sent、user、task_id），通过信号触发任务调度：

# tasks.py
from celery import shared_task
from django.utils import timezone
from .models import ScheduledEvent
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

@shared_task
def send_timed_event(event_id):
    event = ScheduledEvent.objects.get(id=event_id)
    if not event.is_sent and event.send_time <= timezone.now():
        # 这里根据推送方式选择逻辑（WebSocket/推送通知）
        # 示例：通过Channels发送WebSocket消息
        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)(
            f'user_{event.user.id}',
            {'type': 'push_event', 'data': event.payload}
        )
        
        event.is_sent = True
        event.save()

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import ScheduledEvent
from .tasks import send_timed_event

@receiver(post_save, sender=ScheduledEvent)
def schedule_event_task(sender, instance, created, **kwargs):
    if created and not instance.is_sent:
        task = send_timed_event.apply_async(args=[instance.id], eta=instance.send_time)
        instance.task_id = task.id
        instance.save()

# 若需修改定时时间，取消旧任务并重新调度
def update_event_schedule(event_id, new_send_time):
    event = ScheduledEvent.objects.get(id=event_id)
    if event.task_id:
        from celery.task.control import revoke
        revoke(event.task_id, terminate=True)
    new_task = send_timed_event.apply_async(args=[event.id], eta=new_send_time)
    event.send_time = new_send_time
    event.task_id = new_task.id
    event.save()

1.3 数据库模型示例 #

# models.py
from django.db import models
from django.contrib.auth.models import User

class ScheduledEvent(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    send_time = models.DateTimeField(default=timezone.now() + timezone.timedelta(hours=1))
    payload = models.JSONField()  # 存储携带的事件数据
    is_sent = models.BooleanField(default=False)
    task_id = models.CharField(max_length=255, null=True, blank=True)  # 存储Celery任务ID
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ['send_time']

2. 后端向前端推送的两种方式 #

方式1：WebSocket实时推送（前台在线场景） #

用 Django Channels 实现WebSocket服务，适合APP前台运行时接收实时数据：

后端配置

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class EventConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user_id = self.scope['url_route']['kwargs']['user_id']
        self.group_name = f'user_{self.user_id}'
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.group_name, self.channel_name)

    async def push_event(self, event):
        await self.send(text_data=json.dumps(event['data']))

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

websocket_urlpatterns = [
    re_path(r'ws/event/(?P<user_id>\d+)/$', consumers.EventConsumer.as_asgi()),
]

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from . import routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": URLRouter(routing.websocket_urlpatterns),
})

React Native 端实现

用 react-native-websocket 库连接：

import React from 'react';
import { View, Text, Alert } from 'react-native';
import WebSocket from 'react-native-websocket';

const EventListener = ({ userId }) => {
  return (
    <View>
      <WebSocket
        url={`ws://your-backend-domain/ws/event/${userId}/`}
        onMessage={(event) => {
          const data = JSON.parse(event.data);
          Alert.alert('收到定时事件', JSON.stringify(data));
          // 处理业务逻辑
        }}
        onOpen={() => console.log('WebSocket连接成功')}
        onError={(err) => console.error('WebSocket错误:', err)}
        reconnect
      />
    </View>
  );
};

export default EventListener;

方式2：推送通知（后台离线场景） #

用FCM（Android）/APNs（iOS）实现后台推送，适合APP在后台或离线时接收：

后端实现

用 pyfcm 调用FCM API：

# tasks.py 中修改 send_timed_event 函数
from pyfcm import FCMNotification

push_service = FCMNotification(api_key="你的FCM服务器密钥")

def send_timed_event(event_id):
    event = ScheduledEvent.objects.get(id=event_id)
    if not event.is_sent and event.send_time <= timezone.now():
        # 从数据库获取用户的设备Token（RN端上传）
        device_token = event.user.devicetoken.token
        push_service.single_device_data_message(
            registration_id=device_token,
            message_title="定时事件提醒",
            data_message=event.payload
        )
        event.is_sent = True
        event.save()

React Native 端实现

集成 react-native-firebase 处理推送：

import messaging from '@react-native-firebase/messaging';

// 请求推送权限并上传设备Token
async function setupPushNotification() {
  const authStatus = await messaging().requestPermission();
  if (authStatus === messaging.AuthorizationStatus.AUTHORIZED) {
    const token = await messaging().getToken();
    // 上传Token到后端
    fetch('https://your-backend-domain/api/save-token/', {
      method: 'POST',
      headers: {'Content-Type': 'application/json'},
      body: JSON.stringify({token, user_id: 123})
    });
  }
}

// 监听前台推送
messaging().onMessage(remoteMessage => {
  Alert.alert('定时事件', JSON.stringify(remoteMessage.data));
});

// 监听后台推送
messaging().setBackgroundMessageHandler(remoteMessage => {
  console.log('后台收到事件:', remoteMessage.data);
});

3. 服务启动命令 #

• 启动Celery Worker：

  celery -A your_project worker --loglevel=info
  

• 启动Celery Beat（若用数据库动态调度）：

  celery -A your_project beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler
  

• 启动Django ASGI服务（WebSocket需要）：

  daphne your_project.asgi:application
  

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