我正在使用Celery 4.1.0在Django 1.10.3中构建简单的等待列表应用程序.
我有以下基本任务:
@shared_task
def start_user_counter():
logging.info('Task executed @ {}'.format(datetime.datetime.utcnow()))
# This task is executed when user reaches the Top of the queue.
# Send email, perform other stuff in here ...
@shared_task
def update_queue():
curr_time = datetime.datetime.utcnow()
logging.info('Task called @ {}'.format(curr_time))
time_to_exec = curr_time + datetime.timedelta(seconds=10)
# Here, perform checks if task already exists in Redis
# if it does not exist - create a new one and store it to Redis
# if it does exist - update task's ETA.
task_id = start_user_counter.apply_async(eta=time_to_exec)
logging.info('Task ID: {}'.format(task_id))
# ...
update_queue.delay()
每个任务代表等待列表中的一个用户.当新用户被假定从等待名单中删除时(他在ETA上达到顶部),将为新用户分配ETA.但是,每个用户也有可能加快他到达等待名单顶部的时间.
问题:我如何更新现有任务的ETA,以便它比最初预期更早执行?
最佳答案 我设法解决了这个问题.我的解决方案是使用Redis创建
sorted set.对于与该组中的每个用户条目相关联的得分值,我使用表示将用户添加到等待列表中的时间的时间戳.这有助于我让用户在正确的订货人中等候名单.
我还使用Redis hash来存储celery.result.AsyncResult.id
,这是我在使用notify_user.apply_async((self.id,),eta = eta).id创建芹菜任务后立即收到的(请参阅下文).
然后,每当我需要更新任务的ETA时,我必须通过调用AsyncResult.revoke()
来忽略任务,就像这个AsyncResult(self.get_task_id()).revoke(). AsyncResult(self.get_task_id())将返回与我从调用self.get_task_id()获得的id相关联的查询任务状态.在此AsyncResult实例上调用.revoke()会使任何接收任务或已保留任务的worker忽略它.
这将允许我使用新的ETA创建全新的任务,其ID将再次存储在Redis中的相同用户记录中,从而覆盖旧的id值.
我的代码示例特定于我的情况,但底线是:
>创建一个全新的任务并将其celery.result.AsyncResult.id
存储在某处(即self.task_id = T.apply_async((args,),eta = eta).id).
>如果您的新ETA依赖于以前ETA商店的价值,那么某个地方也是如此(即self.eta = eta)
>使用AsyncResult(task_id)创建查询任务状态的实例,并忽略此任务调用.revoke()方法. (即AsyncResult(self.task_id).revoke()
>计算新的ETA并使用新的ETA创建新任务(即self.task_id = T.apply_async((args,),eta = new_eta).id)
#utils.py
import datetime as dt
import redis
from django.conf import settings
from celery.result import AsyncResult
from .tasks import notify_candidate
KEY_DATA = 'user:data'
KEY_QUEUE = 'user:queue'
TIME_DELTA = 'time_delta'
TASK_ID = 'task_id'
WAITING_TIME = 14 * 24 * 60 * 60 # 14 days by default
r = redis.StrictRedis(host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
db=settings.REDIS_DB)
class UserEntry(object):
def __init__(self, user_id):
self.id = user_id
# dynamically creates string for each user that will be later used
# as a key for hash in our Redis storage
self.user_key = '{}:{}'.format(KEY_DATA, user_id)
self.create_or_update()
def create_or_update(self, data=None):
"""
Set up new user entry.
:return: None
"""
if self.exist():
# data exist for user with user_id - update it
r.hmset(self.user_key, data)
else:
# this is a new user - create new entry for this user
self.add_user()
eta = dt.datetime.utcfromtimestamp(self.get_score())
task_id = notify_user.apply_async((self.id,), eta=eta).id
r.hmset(self.user_key, {TASK_ID: task_id})
def add_user(self):
"""
Appends user's ID to the end of the queue.
:return: None
"""
if self.get_index():
# if user entry exits simulate NX option of zadd command -
# Don't update already existing elements. Always add new elements.
return
# use UTC timestamp as score
utc_time = dt.datetime.utcnow()
score = int(utc_time.timestamp()) + WAITING_TIME
r.zadd(KEY_QUEUE, score, self.id)
def get_score(self):
"""
Gets user's score (current ETA).
:return: timestamp representing value of user's ETA
"""
return r.zscore(KEY_QUEUE, self.id)
def get_index(self):
"""
Gets user's position in the queue.
:return: 0-based index value representing user's position in the queue
"""
return r.zrank(KEY_QUEUE, self.id)
def get_task_id(self):
"""
Helper method to get task ID for the user
:return: value of user task's ID
"""
return r.hget(self.user_key, TASK_ID).decode('ascii')
def set_score(self, score_delta):
"""
Move user up in the queue by score value.
:param score_delta: number of seconds by which user's
score (curernt ETA) will be decremented
:return: timestamp representing user's new score (ETA)
"""
r.zincrby(KEY_QUEUE, self.id, score_delta)
def exist(self):
"""
Helper method used to define whether user exists in queue
:return: dict of the hash’s name/value pairs if data entry exist
"""
return r.hgetall(self.user_key)
def bump(self):
"""
Move user up in the queue
:return: None
"""
if not self.exist():
return
# remove current task associated with the user
AsyncResult(self.get_task_id()).revoke()
# we need to decrement ETA, thus *(-1)
# here I make time_delta equal to 1 day or 1 * 24 * 60 * 60 seconds
time_delta = WAITING_TIME / 14 * -1
self.set_score(time_delta)
new_eta = dt.datetime.utcfromtimestamp(time_delta)
task_id = notify_user.apply_async((self.id,), eta=new_eta).id
self.create_or_update({TASK_ID: task_id})
#tasks.py
import datetime
import logging
from celery import shared_task
@shared_task
def notify_user(user_id):
logging.info('Task executed @ {}'.format(datetime.datetime.utcnow()))
loging.info('UserID: {}'.format(user_id))
# This task is executed when user reaches the Top of the queue.
# Send email, perform other stuff in here ...
#models.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .utils import UserEntry
@receiver(post_save, sender=MyUser)
def create_user_entry_in_waiting_list(sender, instance=None, created=False, **kwargs):
if created:
# create user entry in the waiting_list
user_id = instance.id
UserEntry(user_id)