任務中的芹菜任務

Question

我在我的應用程序中將芹菜與Dropbox結合在一起，並且我可以讓用戶在連接了Dropbox的情況下保存自己的照片。

我寫了一段代碼，但我擔心這可能會導致無限循環，從而導致系統死機。

我使用的API只允許在一次提供60張照片，然後爲您提供分頁。

這裏是我的tasks.py文件的副本 - 這實際上工作正常，但我想檢查我做的是正確的事情，而不是太多影響系統。

class DropboxUsers(PeriodicTask): 
    run_every = timedelta(hours=4) 

    def run(self, **kwargs): 
     logger = self.get_logger(**kwargs) 
     logger.info("Collecting Dropbox users") 

     dropbox_users = UserSocialAuth.objects.filter(provider='dropbox') 
     for db in dropbox_users: 
      ... 
      ... 
      ... 
      sync_images.delay(first, second, third_argument) 
     return True 


@task(ignore_result=True) 
def sync_images(token, secret, username): 
    """docstring for sync_images""" 
    logger = sync_images.get_logger() 
    logger.info("Syncing images for %s" % username) 
    ... 
    ... 
    ... 
    ... 
    feed = api.user_recent_media(user_id='self', count=60) 
    images = feed[0] 
    pagination = feed[1] 
    for obj in images: 
     ### STORE TO DROPBOX 
     ... 
     ... 
     ... 
     response = dropbox.put_file(f, my_picture, overwrite=True) 
    ### CLOSE DB SESSION 
    sess.unlink() 
    if pagination: 
     store_images.delay(first, second, third, fourth_argument) 

@task(ignore_result=True) 
def store_images(token, secret, username, max_id): 
    """docstring for sync_images""" 
    logger = store_images.get_logger() 
    logger.info("Storing images for %s" % username) 
    ... 
    ... 
    ... 
    ... 
    feed = api.user_recent_media(user_id='self', count=60, max_id=max_id) 
    images = feed[0] 
    try: 
     pagination = feed[1] 
    except: 
     pagination = None 
    for obj in images: 
     ### STORE TO DROPBOX 
     ... 
     ... 
     ... 
     response = dropbox.put_file(f, my_picture, overwrite=True) 
    ### CLOSE DB SESSION 
    sess.unlink() 
    if pagination: 
     ### BASICALLY RESTART THE TASK WITH NEW ARGS 
     store_images.delay(first, second, third, fourth_argument) 

    return True 

您的專業知識非常感謝。

Answer 1

我沒有看到任何重大問題。我還實施了一項任務啓動另一項任務的系統。

有一段時間，我在服務器重啓時遇到了芹菜重複任務的問題。我編寫了一個裝飾器，用於封裝使用高速緩存後端的任務，以確保具有相同參數的相同任務不會經常運行。可能會有用作對衝你的無限循環。

from django.core.cache import cache as _djcache 
from django.utils.functional import wraps 

class cache_task(object): 

    """ Makes sure that a task is only run once over the course of a configurable 
    number of seconds. Useful for tasks that get queued multiple times by accident, 
    or on service restart, etc. Uses django's cache (memcache) to keep track.""" 

    def __init__(self, seconds=120, minutes=0, hours=0): 
     self.cache_timeout_seconds = seconds + 60 * minutes + 60 * 60 * hours 

    def __call__(self, task): 
     task.unsynchronized_run = task.run 
     @wraps(task.unsynchronized_run) 
     def wrapper(*args, **kwargs): 
      key = sha1(str(task.__module__) + str(task.__name__) + str(args) + str(kwargs)).hexdigest() 
      is_cached = _djcache.get(key) 
      if not is_cached: 
       # store the cache BEFORE to cut down on race conditions caused by long tasks 
       if self.cache_timeout_seconds: 
        _djcache.set(key, True, self.cache_timeout_seconds) 
       task.unsynchronized_run(*args, **kwargs) 
     task.run = wrapper 
     return task 

用法：

@cache_task(hours=2) 
@task(ignore_result=True) 
def store_images(token, secret, username, max_id): 
    ...