如何用Python实现Celery定时任务24小时内仅触发一次消息发送？

核心思路是记录每个对象的上次消息发送时间，通过持久化存储保存这个时间，每次任务执行时同时判断两个条件：

• 对象的shussui为True
• 距离上次发送时间已超过24小时（或从未发送过）

满足条件才发送消息，并更新上次发送时间。

方案1：用Redis存储发送记录（轻量高效） #

Redis适合存储这类键值对型的时间记录，操作简单且性能高。

步骤： #

1. 安装Redis客户端：

pip install redis

2. 修改任务代码：

import redis
from datetime import datetime, timedelta

# 初始化Redis连接，根据你的Redis配置调整参数
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def your_celery_task(object_id):
    # 替换成你实际获取对象的逻辑
    object = get_your_object_by_id(object_id)
    
    if object.shussui:
        # 用对象ID作为唯一键，区分不同对象的发送记录
        key = f"last_sent:{object.id}"
        last_sent_str = redis_client.get(key)
        
        need_send = False
        if not last_sent_str:
            # 从未发送过，直接触发发送
            need_send = True
        else:
            # 解析上次发送时间，计算时间差
            last_sent_time = datetime.fromisoformat(last_sent_str)
            if datetime.now() - last_sent_time >= timedelta(hours=24):
                need_send = True
        
        if need_send:
            # 执行消息发送
            client.conversation_start({
                'channelId': 'x',
                'to': 'user',
                'type': 'text',
                'content': {
                    'text': 'body message'
                }
            })
            # 更新Redis中的发送时间
            redis_client.set(key, datetime.now().isoformat())

方案2：用数据库字段存储（适合已有关系型数据库的场景） #

如果你的对象本身存储在关系型数据库（比如Django Model、SQLAlchemy模型），可以直接给模型新增字段记录发送时间。

步骤： #

1. 修改模型（以Django为例）：

from django.db import models

class YourModel(models.Model):
    shussui = models.BooleanField(default=False)
    # 新增字段，允许为空，记录上次发送时间
    last_sent = models.DateTimeField(null=True, blank=True)
    # 其他原有字段...

执行数据库迁移命令，更新表结构。

2. 修改任务代码：

from datetime import datetime, timedelta

def your_celery_task(object_id):
    object = YourModel.objects.get(id=object_id)
    
    if object.shussui:
        need_send = False
        if not object.last_sent:
            need_send = True
        else:
            if datetime.now() - object.last_sent >= timedelta(hours=24):
                need_send = True
        
        if need_send:
            client.conversation_start({
                'channelId': 'x',
                'to': 'user',
                'type': 'text',
                'content': {
                    'text': 'body message'
                }
            })
            # 更新数据库中的发送时间
            object.last_sent = datetime.now()
            object.save()

注意事项 #

• 确保Celery Worker能正常连接到Redis或数据库，避免连接失败导致记录丢失。
• 分布式部署时，要保证所有Worker访问同一个Redis/数据库实例。
• 建议用UTC时间（比如datetime.utcnow()）计算时间差，避免时区差异引发的判断错误。

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