python - Comment ajouter dynamiquement des tâches pendant l'exécution du planificateur Celery Beat?

Question

J'ai essayé Django-celery-beat. L'ajout de tâches en arrière-plan d'administration peut permettre un ajout dynamique de tâches
Mais cela ne prendra effet qu'après le redémarrage de Celery Beat. Existe-t-il un autre moyen d'essayer ?

Answer 1

Ne peut pas être ajouté dynamiquement, le rythme doit être redémarré.

demandez à répondre à la raison #3493

Answer 2

Il y a une idée que vous pouvez envisager. J'essaie actuellement cette méthode et je suis en train d'essayer de traverser la rivière en palpant les pierres. Celery prend en charge les tâches planifiées, mais cela ne répond pas à mes besoins. J'ai besoin d'ajouter dynamiquement des tâches planifiées comme crontab sous Linux. J'ai également regardé Django-celery-beat parce que j'utilise Flask, j'ai trouvé que cela ne valait pas la peine d'être référencé pour l'implémentation. , donc j'ai été Après avoir lu la documentation et recherché des informations, j'ai finalement trouvé un moyen. La fonction apply_async de Celery est très utile. Elle a un paramètre eta. Son utilisation simplifiée est un compte à rebours, mais la puissance de eta est énorme car elle n'accepte que. objets datetime , par exemple, si vous donnez une tâche à exécuter le 2017-05-02 20:0:0, vous pouvez l'utiliser comme ceci :

job.apply_async(args=args, kwarg=kwargs, eta=datetime(2017,5,2,20,0,0))

Est-il rarement utilisé ? Si j'ai une tâche qui doit être exécutée à huit heures tous les soirs, je peux utiliser ce paramètre eta pour y parvenir. Le pseudo code est le suivant :

时间规则 = '每天晚上八点执行'
第一次调用任务，先计算最近的执行时间，作为eta的参数，调用apply_async函数，
然后第一次任务执行成功，得到上次任务的eta参数值，在天的值上加一，然后把新的执行时间作为eta的参数再次调用apply_async函数，这里省略了很多判断，自行脑补。
循环往复，是不是一直按每天晚上八点执行。

Un point très important ici est de savoir comment calculer l'heure d'exécution suivante lorsque la tâche est exécutée avec succès. La méthode est la suivante

.

class MyTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        print 'task done: {0}'.format(retval)
        return super(MyTask, self).on_success(retval, task_id, args, kwargs)
        
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print 'task fail, reason: {0}'.format(exc)
        return super(MyTask, self).on_failure(exc, task_id, args, kwargs, einfo)
        
@app.task(base=MyTask)
def add(x, y):
    return x + y

Il fournit des fonctions pour le succès et l'échec de l'exécution des tâches. Il suffit de le réécrire sur cette base. Je ne parle que de la partie principale. Il existe de nombreuses façons de le faire spécifiquement,

.