Django中Redis Pub/Sub与多线程优化邮件发送的问题求助

问题1：Pub/Sub消息监听异常（始终收到{"data":1}） #

原因分析 #

这种情况大概率是发布端错误地将Redis publish方法的返回值当作消息发送导致的。Redis的publish方法返回当前订阅该频道的客户端数量（比如仅1个订阅端时返回1），如果发布逻辑误把这个返回值发布出去，就会出现订阅端收到{"data":1}的情况。另外，也可能是订阅端未正确过滤Redis的订阅确认消息（订阅成功时会返回subscribe类型消息，data字段是订阅的频道数）。

解决方案 #

• 检查发布端代码：确保发布的是实际邮件任务数据（含收件人、标题、内容的字典），而非publish方法的返回值。示例正确逻辑：

  import json
  from redis import Redis
  
  redis_client = Redis()
  email_task = {"to": "user@example.com", "subject": "Test", "content": "Hello"}
  # 发布序列化后的任务数据
  redis_client.publish("email_campaign", json.dumps(email_task))
  

• 订阅端过滤无效消息：监听时只处理message类型的消息，正确解析data字段：

  pubsub = redis_client.pubsub()
  pubsub.subscribe("email_campaign")
  for message in pubsub.listen():
      # 跳过订阅确认等系统消息，只处理实际任务
      if message["type"] == "message":
          task_data = json.loads(message["data"])
          handle_email_task(task_data)
  

问题2：send_email函数静默失败导致死循环 #

原因分析 #

• fail_silently=False仅控制Django send_mail是否抛出SMTP异常，但如果send_email工具函数中存在自定义try-except块吞掉了异常，就会导致异常无法触发，线程重复执行失败任务，最终拖垮请求响应。
• 也可能是SMTP配置错误（如EMAIL_HOST、认证信息错误），但异常被未正确处理的代码掩盖。

解决方案 #

• 检查./campaigns/utils.py中的send_email实现：移除不必要的try-except块，确保Django send_mail的异常能向上传递。示例正确写法：

  from django.core.mail import send_mail
  
  def send_email(task_data):
      send_mail(
          subject=task_data["subject"],
          message=task_data["content"],
          from_email="your@example.com",
          recipient_list=[task_data["to"]],
          fail_silently=False,
      )
  

• 在线程任务逻辑中添加异常捕获与日志：

  import logging
  
  logger = logging.getLogger(__name__)
  
  def handle_email_task(task_data):
      try:
          send_email(task_data)
          logger.info(f"Email sent to {task_data['to']}")
      except Exception as e:
          logger.error(f"Failed to send email to {task_data['to']}: {str(e)}")
          # 终止当前任务，避免死循环
          return
  

• 验证SMTP配置：直接在Django Shell中调用send_mail测试，确认配置能正常发送邮件，排除基础配置问题。

基于Pub/Sub与多线程的邮件发送优化方案 #

1. 替换Pub/Sub为Redis Stream #

Redis Pub/Sub是无持久化的广播模型，订阅端下线后消息会丢失。改用Redis Stream可实现持久化消息队列，还支持消费者组、消息确认机制，避免消息丢失和重复消费。

2. 使用线程池管理线程 #

手动创建线程易导致资源耗尽，改用concurrent.futures.ThreadPoolExecutor可高效控制并发度：

from concurrent.futures import ThreadPoolExecutor

# 初始化线程池，设置最大并发数
executor = ThreadPoolExecutor(max_workers=5)

# 消息监听逻辑中提交任务到线程池
for message in stream.listen():
    task_data = json.loads(message["data"])
    executor.submit(handle_email_task, task_data)

3. 添加任务超时与重试机制 #

用tenacity库实现带指数退避的重试，避免临时网络问题导致任务失败，同时设置最大重试次数防止死循环：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def send_email(task_data):
    send_mail(...)

4. 任务状态追踪 #

在Redis中记录每个邮件任务的状态（待发送、发送中、成功、失败），方便后续排查和统计：

def handle_email_task(task_data):
    task_id = task_data["task_id"]
    redis_client.set(f"email_task:{task_id}", "processing")
    try:
        send_email(task_data)
        redis_client.set(f"email_task:{task_id}", "success")
    except Exception as e:
        redis_client.set(f"email_task:{task_id}", f"failed:{str(e)}")

5. 批量处理邮件 #

将多个收件人打包成一个任务消息，减少消息队列的消息量，提升处理效率：

# 发布端批量打包任务
batch_task = {"recipients": ["user1@example.com", "user2@example.com"], "subject": "Batch Test", "content": "Hello"}
redis_client.xadd("email_campaign_stream", {"data": json.dumps(batch_task)})

# 订阅端批量处理
def handle_batch_task(batch_data):
    for recipient in batch_data["recipients"]:
        executor.submit(send_email, {"to": recipient, "subject": batch_data["subject"], "content": batch_data["content"]})

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