一文教你实战构建消息通知系统Django

为了实现一个消息通知系统，我们可以使用Django的信号和队列来实现。以下是一个简化的例子：

首先，安装Django以及celery和redis（或者其他你喜欢的消息队列）：

pip install django celery redis

然后，设置你的Django项目以使用Celery：

# your_project/__init__.py
from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app
 
__all__ = ['celery_app']
 
# your_project/celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
 
app = Celery('your_project')
 
app.config_from_object('django.conf:settings', namespace='CELERY')
 
app.autodiscover_tasks()
 
@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

接下来，定义一个信号和一个处理该信号的异步任务：

# your_app/models.py
from django.db import models
from django.dispatch import receiver
from django.db.models.signals import post_save
from django.utils.translation import gettext_lazy as _
from .tasks import send_notification
 
class Notification(models.Model):
    message = models.CharField(max_length=255)
    # other fields...
 
@receiver(post_save, sender=Notification)
def notify_user(sender, instance, created, **kwargs):
    if created:
        send_notification.delay(instance.message)
 
# your_app/tasks.py
from __future__ import absolute_import, unicode_literals
from celery import shared_task
 
@shared_task
def send_notification(message):
    # 发送通知逻辑，例如发送邮件或者推送通知
    print(f"Sending notification: {message}")
    # send_email_notification(message) 或 send_push_notification(message)

在你的Django设置中配置Celery：

# your_project/settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TIMEZONE = 'Asia/Shanghai'

最后，确保你的 Celery 工人正在运行：

celery -A your_project worker -l info

当你创建一个Notification对象时，post_save信号会触发，然后异步任务send_notification会被调用来发送通知。这个例子使用了打印语句来模拟发送通知，你应该替换它以实现实际的通知发送逻辑。