在使用 django_celery_beat 设置的 Django 视图中使用 Celery 定期任务输出，并使用 Redis 设置缓存

Posted 2023-02-24

【中文标题】在使用 django_celery_beat 设置的 Django 视图中使用 Celery 定期任务输出，并使用 Redis 设置缓存

【英文标题】：Use Celery periodic tasks output in Django views set up with django_celery_beat and Cachine with Redis

【发布时间】：2020-10-26 01:26:12

【问题描述】：

我正在尝试使用 Celery 在我的一个模型上执行相当耗时的算法。 目前在我的 home.tasks.py 我有：

@shared_task(bind=True)
def get_hot_posts():
    return Post.objects.get_hot()
    
    
@shared_task(bind=True)
def get_top_posts():
    pass

在我的Post 对象模型管理器中我有：

def get_hot(self):
    
    qs = (
        self.get_queryset()
        .select_related("author")
    )
    
    qs_list = list(qs)
    sorted_post = sorted(qs_list, key=lambda p: p.hot(), reverse=True)
    
    return sorted_post

返回热门帖子的列表对象。

我使用 django_celery_beat 来设置周期性任务。我在 settings.py 中配置的

CELERY_BEAT_SCHEDULE = 
    'update-hot-posts': 
        'task':'get_hot_posts',
        'schedule': 3600.0
    ,
    'update-top-posts': 
        'task':'get_top_posts',
        'schedule': 86400
    

如果我可以在 Celery 任务中对我的模型执行任何功能，我不会这样做，但我的目的是每 1 小时计算一次热门帖子，然后在我的一个视图中简单地使用它。我该如何实现这一点，我无法找到如何获取该任务的输出并在我的视图中使用它以便在我的模板中呈现它。

提前致谢！

编辑

我现在正在缓存结果：

settings.py：

CACHES = 
    "default": 
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "OPTIONS": 
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "IGNORE_EXCEPTIONS": True,
            
        
    

CACHE_TTL = getattr(settings, 'CACHE_TTL', DEFAULT_TIMEOUT)

@shared_task(bind=True)
def get_hot_posts():
    hot_posts = Post.objects.get_hot()
    cache.set("hot_posts", hot_posts, timeout=CACHE_TTL)

但是，当访问我视图中的对象时，它返回 None，看来我的任务不起作用。

@login_required
def hot_posts(request):
    posts = cache.get("hot_posts")
    context =  'posts':posts, 'hot_active':'-active'
    return render(request, 'home/homepage/home.html', context)

如何检查我的任务是否正常运行？它实际上正在工作并缓存查询集函数。

编辑：settings.py中的配置：

BROKER_URL = 'redis://localhost:6379'
BROKER_TRANSPORT = 'redis'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_BEAT_SCHEDULE = 
    'update-hot-posts': 
        'task':'get_hot_posts',
        'schedule': 3600.0
    ,
    'update-top-posts': 
        'task':'get_top_posts',
        'schedule': 86400.0
    ,
    'tester': 
        'task':'tester',
        'schedule': 60.0
    

当我转到我的视图并且cache.get 返回 None 时，我没有看到和结果，我认为我的任务没有运行但我找不到原因。

这就是我运行我的工作人员时发生的情况：

celery -A register worker -E --loglevel=info

 -------------- [e-mail] v4.4.6 (cliffs)
--- ***** ----- 
-- ******* ---- Darwin-16.7.0-x86_64-i386-64bit 2020-07-06 01:46:36
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         register:0x10f3da050
- ** ---------- .> transport:   redis://localhost:6379//
- ** ---------- .> results:     redis://localhost:6379/
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: ON
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . home.tasks.get_hot_posts
  . home.tasks.get_top_posts
  . home.tasks.tester

[2020-07-06 01:46:38,449: INFO/MainProcess] Connected to redis://localhost:6379//
[2020-07-06 01:46:38,500: INFO/MainProcess] mingle: searching for neighbors
[2020-07-06 01:46:39,592: INFO/MainProcess] mingle: all alone
[2020-07-06 01:46:39,650: INFO/MainProcess] [e-mail] ready.

也用于我使用的启动节拍：

  celery -A register beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

【问题讨论】：

您需要以某种方式缓存数据。您可以通过数据模型使用 Redis 或您的数据库来做到这一点。

现在添加了缓存，但是我现在找不到如何在我的视图中实现缓存对象。我该如何实现呢？ @schillingt

【参考方案1】：

我的建议是您更改模型并使其可标记。也许是这样：https://django-taggit.readthedocs.io/

完成后，您可以修改计算热门帖子的 celery 作业。计算出新的热门帖子后，您可以从所有现有帖子中删除所有“热门”标签，然后用“热门”标签标记新的热门帖子。

然后您的视图代码可以简单地过滤带有热标签的帖子。

编辑

如果您想确保您的代码确实在执行，您可以使用一些扩展来执行此操作。例如，django-celery-results 后端将在数据库中存储您 @shared_task 返回的任何数据（如果那是您的消息编码，通常是 JSON）以及时间戳，甚至可能是输入参数。因此，您可以查看您的任务是否按预期运行。

https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html#django-celery-results-using-the-django-orm-cache-as-a-result-backend

你也可以考虑使用 django-celery-beat 来确保你有一个很好的视觉方式来通过 django 管理员查看工作计划

https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html#django-celery-beat-database-backed-periodic-tasks-with-admin-interface

编辑 2

如果您要使用数据库调度程序（强烈推荐！），那么您需要登录管理员并按照您想要的时间表添加任务。

https://pinoylearnpython.com/wp-content/uploads/2019/04/Django-Celery-Beat-on-Admin-Site-Pinoy-Learn-Python-1024x718.jpg

编辑 3

在你的 settings.py 中

CELERY_BEAT_SCHEDULE = 
    'update-hot-posts': 
        'task':'get_hot_posts',
        'schedule': 3600.0
    ,
    'update-top-posts': 
        'task':'get_top_posts',
        'schedule': 86400.0
    ,
    'tester': 
        'task':'tester',
        'schedule': 60.0
    

那里的第三个任务称为tester，它应该每 60 秒运行一次。我在您的任务中的任何地方都看不到这一点。因为您试图安排一个未在任何地方定义的任务，因为 @shared_task celery 会感到困惑并给您有关 tester 的错误消息。

【讨论】：

谢谢，这是个好主意，但是，我现在不确定我的任务是否正常运行，我已经运行了统计信息以查看它是否正常工作并添加了代码供您参考.

所以我不能从设置文件中添加任务？就像我现在做的那样。

你可以！但是你不应该使用数据库调度程序，它们不会出现在管理员中。由你决定。因为您使用的是数据库调度程序（ --scheduler 参数），所以它正在查看数据库以查找要调度的任务，但没有找到，因此没有任何东西在运行。

好的，我明白了！所以以我目前的配置，我应该简单地运行celery -A register beat -l INFO?然后也解雇工人？

是的！或者登录到管理员并以这种方式添加任务，然后使用数据库调度程序。我更喜欢这种方法，因为它可以让我把日程安排的责任交给我以外的人。但最终这与让某些东西先工作相比是微不足道的。