Django后端批量发邮件遇Celery连接问题及WebSocket需求咨询

一、解决Celery+Redis连接拒绝([Errno 111])问题 #

1. 检查Redis服务状态 #

• 执行redis-cli ping，返回PONG说明Redis正常运行，否则启动服务：
  • Linux：systemctl start redis-server
  • Windows：在Redis安装目录运行redis-server.exe
• 打开redis.conf，注释bind 127.0.0.1或改为bind 0.0.0.0，同时设置protected-mode no（生产环境需搭配密码认证）

2. 验证Celery配置 #

• 确保settings.py中的broker和result backend URL正确：

  CELERY_BROKER_URL = 'redis://localhost:6379/0'
  CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  

• 启动Celery worker时指定正确项目模块：celery -A your_project_name worker --loglevel=info

3. 修复多收件人处理逻辑 #

• 检查视图循环代码，确保所有收件人都被处理，示例修正逻辑：

  # views.py
  from celery import shared_task
  
  @shared_task
  def send_invitation_email(recipient_email):
      # 邮件发送逻辑（调用SMTP服务）
      pass
  
  def send_invitations(request):
      recipients = [email.strip() for email in request.POST.get('recipients').split(',')]
      for email in recipients:
          Invitation.objects.create(email=email, status='pending')
          send_invitation_email.delay(email)
      return JsonResponse({'status': 'success'})
  

4. 排查端口与防火墙 #

• 检查Redis默认端口6379是否被防火墙拦截，开放端口：
  • Linux：ufw allow 6379
  • Windows：在防火墙高级设置中添加入站规则允许6379端口

二、WebSocket搭建实现状态实时更新 #

1. 配置Django Channels（以Django生态为例） #

• 安装依赖：pip install channels channels-redis
• 在settings.py中添加配置：

  INSTALLED_APPS = [
      # 其他应用
      'channels',
  ]
  
  ASGI_APPLICATION = 'your_project.asgi.application'
  
  CHANNEL_LAYERS = {
      'default': {
          'BACKEND': 'channels_redis.core.RedisChannelLayer',
          'CONFIG': {'hosts': [('localhost', 6379)]},
      },
  }
  

2. 创建WebSocket消费者 #

• 在应用目录新建consumers.py：

  import json
  from channels.generic.websocket import AsyncWebsocketConsumer
  
  class InvitationStatusConsumer(AsyncWebsocketConsumer):
      async def connect(self):
          self.user_group = f'invitation_{self.scope["user"].id}'
          await self.channel_layer.group_add(self.user_group, self.channel_name)
          await self.accept()
  
      async def disconnect(self, close_code):
          await self.channel_layer.group_discard(self.user_group, self.channel_name)
  
      async def push_status(self, event):
          await self.send(text_data=json.dumps(event['data']))
  

3. 配置WebSocket路由 #

• 新建routing.py：

  from django.urls import re_path
  from . import consumers
  
  websocket_urlpatterns = [
      re_path(r'ws/invitation/status/$', consumers.InvitationStatusConsumer.as_asgi()),
  ]
  

• 修改项目根目录asgi.py：

  import os
  from django.core.asgi import get_asgi_application
  from channels.routing import ProtocolTypeRouter, URLRouter
  from channels.auth import AuthMiddlewareStack
  import your_app.routing
  
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
  
  application = ProtocolTypeRouter({
      "http": get_asgi_application(),
      "websocket": AuthMiddlewareStack(URLRouter(your_app.routing.websocket_urlpatterns)),
  })
  

4. 在Celery任务中推送状态 #

• 修改Celery任务，发送邮件后推送状态到WebSocket：

  from channels.layers import get_channel_layer
  from asgiref.sync import async_to_sync
  
  @shared_task
  def send_invitation_email(recipient_email, user_id):
      # 执行邮件发送逻辑
      # 更新数据库状态
      invitation = Invitation.objects.get(email=recipient_email)
      invitation.status = 'delivered'
      invitation.save()
  
      # 推送状态到用户的WebSocket组
      channel_layer = get_channel_layer()
      async_to_sync(channel_layer.group_send)(
          f'invitation_{user_id}',
          {
              'type': 'push_status',
              'data': {'email': recipient_email, 'status': 'delivered'}
          }
      )
  

5. 前端WebSocket连接 #

• 在页面中添加监听代码：

  const socket = new WebSocket(`ws://${window.location.host}/ws/invitation/status/`);
  socket.onmessage = function(e) {
      const data = JSON.parse(e.data);
      // 更新页面中对应收件人的状态显示
      document.querySelector(`[data-email="${data.email}"]`).textContent = data.status;
  };
  

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