使用Django Celery按publish_date定时发布Post模型内容的问题

我有一个Post模型，包含publish_date字段。当帖子状态为planned时，需要在publish_date指定时间自动执行发布操作。尝试通过Django的post_save信号调用Celery异步任务实现，但任务始终未执行成功。

相关代码 #

初始models.py

class Post(models.Model):
    STATES = (
        ('draft', 'Draft'),
        ('published', 'Published'),
        ('planned', 'Planned')
    )
    state = models.CharField(choices=STATES, default=STATES[0][0])
    channels = models.ManyToManyField('channel.Channel')
    creator = models.ForeignKey('authentication.User', on_delete=models.SET_NULL, null=True)
    publish_date = models.DateTimeField()
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)


@receiver(post_save, sender=Post)
def reschedule_publish_task(sender, instance, **kwargs):
    # 这里应该设置定时任务，我是这么理解的

尝试的models.py修改版

@receiver(post_save, sender=Post)
def reschedule_publish_task(sender, instance, **kwargs):
    task_id = f"publish_post_{instance.id}"

    if instance.state == 'planned' and instance.publish_date:
        publish_post.apply_async((instance.id,), eta=instance.publish_date, task_id=task_id)

tasks.py

@shared_task
def publish_post(post_id: int) -> None:
    from .models import Post

    post = Post.objects.filter(id=post_id).first()
    if post:
        if post.state == 'planned' and post.publish_date <= now():
            post.state = 'published'
            post.save()

1. 时区配置不一致 #

Celery默认时区可能与Django的时区设置不匹配，导致eta时间计算错误，任务无法在指定时间执行。

• 解决办法：在Celery配置文件（如celery.py）中设置与Django一致的时区：

app.conf.timezone = 'Asia/Shanghai'  # 替换为你Django settings里的TIME_ZONE值

同时修改tasks.py中的时间对比，使用Django时区工具：

from django.utils import timezone

@shared_task
def publish_post(post_id: int) -> None:
    from .models import Post

    post = Post.objects.filter(id=post_id).first()
    if post:
        if post.state == 'planned' and post.publish_date <= timezone.now():
            post.state = 'published'
            post.save()

2. 重复任务未清理 #

当Post实例更新（比如修改publish_date或state）时，旧的定时任务仍在队列中，新任务使用相同task_id会导致冲突，或者旧任务依然执行覆盖新逻辑。

• 解决办法：在信号函数中先取消旧任务，再添加新任务：

from celery.task.control import revoke
from django.utils import timezone

@receiver(post_save, sender=Post)
def reschedule_publish_task(sender, instance, **kwargs):
    task_id = f"publish_post_{instance.id}"
    
    # 强制取消已存在的同名任务
    revoke(task_id, terminate=True, ignore_result=True)
    
    if instance.state == 'planned' and instance.publish_date:
        # 仅当发布时间在未来时才添加任务
        if instance.publish_date > timezone.now():
            publish_post.apply_async((instance.id,), eta=instance.publish_date, task_id=task_id)
    else:
        # 状态非planned时，清理已存在的任务
        revoke(task_id, terminate=True, ignore_result=True)

3. Celery任务未注册 #

确保Celery worker能发现你的publish_post任务。

• 解决办法：在Celery配置中指定任务所在模块：

app.conf.include = ['your_app_name.tasks']  # 替换为你的app名称

同时重启Celery worker，确保任务被重新加载。

4. 信号循环触发问题 #

post_save信号会在每次post.save()时触发，包括任务执行时修改状态的save()操作，导致信号重复执行，可能引发异常或重复添加任务。

• 解决办法：利用信号的created参数或判断操作类型，避免循环：

@receiver(post_save, sender=Post)
def reschedule_publish_task(sender, instance, created=False, **kwargs):
    # 仅在实例更新时处理，或排除任务触发的save操作
    task_id = f"publish_post_{instance.id}"
    revoke(task_id, terminate=True, ignore_result=True)
    
    if instance.state == 'planned' and instance.publish_date:
        if instance.publish_date > timezone.now():
            publish_post.apply_async((instance.id,), eta=instance.publish_date, task_id=task_id)

5. Celery Worker状态检查 #

确认Celery worker是否正常运行，查看日志排查错误：

• 启动worker命令：

celery -A your_project_name worker -l info

查看日志中是否有任务接收、调度或执行的记录，检查是否有报错信息（如数据库连接失败、任务导入错误等）。

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