Exécution de tâches différées avec Redis

J'ai besoin de concevoir un Redis-driven évolutive de la planification des tâches du système.

Exigences:

Plusieurs processus de travail.
De nombreuses tâches, mais de longues périodes d'inactivité sont possibles.
Raisonnable précision de la synchronisation.
Minimum le gaspillage de ressources en cas d'inactivité.
Devrait synchrone Redis API.
Devrait travailler pour le Redis 2.4 (c'est à dire pas de fonctionnalités à venir de la 2.6).
Ne devrait pas utiliser d'autres moyens de RPC que le Redis.

Pseudo-API: schedule_task(timestamp, task_data). Timestamp est un entier de secondes.

Idée de base:

Écouter pour les tâches à venir sur la liste.
Mettre tâches à seaux par horodateur.
Sommeil jusqu'à l'horodateur le plus proche.
Si une nouvelle tâche s'affiche avec horodatage de moins que le plus proche, se réveiller.
Traiter toutes les tâches à venir avec horodatage ≤ maintenant, dans les lots (en supposant que
que l'exécution de la tâche est rapide).
Assurez-vous que le travail simultanées ne traitent pas les mêmes tâches. Dans le même temps, assurez-vous qu'aucune tâche n'est perdu si nous crash lors du traitement.

Jusqu'à présent, je n'arrive pas à comprendre comment intégrer le Redis primitives...

Toute indices?

Remarque qu'il est à la même vieille question: Retards dans l'exécution /la planification avec le Redis? Dans cette nouvelle question, je introduire plus de détails (le plus important, de nombreux travailleurs). Jusqu'à présent je n'étais pas en mesure de comprendre comment l'appliquer vieux réponses ici —ainsi, une nouvelle question.

source d'informationauteur Alexander Gladysh

Voici une autre solution qui s'appuie sur un couple d'autres [1]. Il utilise le redis REGARDER la commande de suppression de la condition de la course sans utiliser lua dans le redis 2.6.

Le schéma de base est:

Utiliser un redis zset pour les tâches planifiées et redis les files d'attente pour les prêts à exécuter les tâches.
Ont un répartiteur sondage la zset et de déplacer des tâches qui sont prêts à courir dans le redis files d'attente. Vous souhaitez peut-être plus de 1 répartiteur pour la redondance, mais vous n'avez probablement pas besoin de beaucoup.
Ont autant de travailleurs que vous voulez qui ne le blocage des pop sur le redis files d'attente.

Je n'ai pas testé 🙂

Les foo créateur d'emplois:

def schedule_task(queue, data, delay_secs):
    # This calculation for run_at isn't great- it won't deal well with daylight
    # savings changes, leap seconds, and other time anomalies. Improvements
    # welcome :-)
    run_at = time.time() + delay_secs

    # If you're using redis-py's Redis class and not StrictRedis, swap run_at &
    # the dict.
    redis.zadd(SCHEDULED_ZSET_KEY, run_at, {'queue': queue, 'data': data})

schedule_task('foo_queue', foo_data, 60)

Le répartiteur(s):

while working:
    redis.watch(SCHEDULED_ZSET_KEY)
    min_score = 0
    max_score = time.time()
    results = redis.zrangebyscore(
        SCHEDULED_ZSET_KEY, min_score, max_score, start=0, num=1, withscores=False)
    if results is None or len(results) == 0:
        redis.unwatch()
        sleep(1)
    else: # len(results) == 1
        redis.multi()
        redis.rpush(results[0]['queue'], results[0]['data'])
        redis.zrem(SCHEDULED_ZSET_KEY, results[0])
        redis.exec()

Les foo travailleur ressemblerait à:

while working:
    task_data = redis.blpop('foo_queue', POP_TIMEOUT)
    if task_data:
        foo(task_data)

[1] Cette solution est basée sur not_a_golfer, l'un à http://www.saltycrane.com/blog/2011/11/unique-python-redis-based-queue-delay/et le redis docs pour les transactions.

5

Vous n'avez pas à spécifier la langue que vous utilisez. Vous avez au moins 3 variantes de faire cela sans écrire une seule ligne de code en Python au moins.
1. Le céleri a une option redis courtier.
  http://celeryproject.org/
2. resque est extrêmement populaire redis tâche de la file d'attente à l'aide de redis.
  https://github.com/defunkt/resque
3. RQ est un simple et petit redis en fonction de la file d'attente qui vise à "prendre les bonnes choses de céleri et resque" et être beaucoup plus simple de travailler avec.
  http://python-rq.org/
Vous pouvez au moins regarder leur design si vous ne pouvez pas les utiliser.

Mais pour répondre à votre question - ce que vous voulez peut être fait avec le redis. En fait, j'ai écrit plus ou moins que dans le passé.

EDIT:
Comme pour la modélisation de ce que vous voulez sur le redis, c'est ce que je ferais:
1. de files d'attente d'une tâche avec un horodatage sera effectuée directement par le client, - vous mettez la tâche en un ensemble trié avec le timestamp comme le score et la tâche de la valeur (voir ZADD).
2. Un répartiteur central se réveille toutes les N secondes, vérifie la première date et l'heure sur ce jeu, et si il y a des tâches prêtes à être exécutées, il pousse la tâche à un "être exécutée MAINTENANT" de la liste. Cela peut être fait avec ZREVRANGEBYSCORE sur le "en attente" ensemble trié, obtenir tous les articles avec le timestamp<=now, de sorte que vous obtenez tous les éléments prêts à la fois. pousser est fait par RPUSH.
3. les travailleurs utilisent BLPOP sur le "pour être exécuté MAINTENANT" liste de le réveiller quand il y a quelque chose à travailler, et faire leur chose. C'est sûr, depuis le redis est mono-thread, et pas de 2 travailleurs prendra jamais la même tâche.
4. une fois terminé, les travailleurs de mettre le résultat dans une file d'attente de réponse, ce qui est vérifié par le répartiteur ou un autre thread. Vous pouvez ajouter un "en attente" seau pour éviter les échecs ou quelque chose comme ça.
de sorte que le code devrait ressembler à quelque chose comme ceci (c'est juste le pseudo-code):

client:
```
ZADD "new_tasks" <TIMESTAMP> <TASK_INFO>
```
répartiteur:
```
while working:
   tasks = ZREVRANGEBYSCORE "new_tasks" <NOW> 0 #this will only take tasks with timestamp lower/equal than now
   for task in tasks:

       #do the delete and queue as a transaction
       MULTI
       RPUSH "to_be_executed" task
       ZREM "new_tasks" task
       EXEC

   sleep(1)
```
Je n'ai pas ajouter à la file d'attente de réponse de la manipulation, mais c'est plus ou moins comme le travailleur:

travailleur:
```
while working:
   task = BLPOP "to_be_executed" <TIMEOUT>
   if task:
      response = work_on_task(task)
      RPUSH "results" response
```
EDit: apatrides atomique répartiteur :
```
while working:

   MULTI
   ZREVRANGE "new_tasks" 0 1
   ZREMRANGEBYRANK "new_tasks" 0 1
   task = EXEC

   #this is the only risky place - you can solve it by using Lua internall in 2.6
   SADD "tmp" task

   if task.timestamp <= now:
       MULTI
       RPUSH "to_be_executed" task
       SREM "tmp" task
       EXEC
   else:

       MULTI
       ZADD "new_tasks" task.timestamp task
       SREM "tmp" task
       EXEC

   sleep(RESOLUTION)
```

Si vous êtes à la recherche de solution prête à l'emploi sur l'île de Java. Redisson est bon pour vous. Il permet de planifier et d'exécuter des tâches (avec cron-expression de soutien) de manière distribuée sur Redisson nœuds utilisation familière ScheduledExecutorService api et basé sur le Redis file d'attente.

Ici est un exemple. D'abord définir une tâche à l'aide de java.lang.Runnable interface. Chaque tâche peut accéder à Redis exemple via injecté RedissonClient objet.

public class RunnableTask implements Runnable {

    @RInject
    private RedissonClient redissonClient;

    @Override
    public void run() throws Exception {
        RMap<String, Integer> map = redissonClient.getMap("myMap");
        Long result = 0;
        for (Integer value : map.values()) {
            result += value;
        }
        redissonClient.getTopic("myMapTopic").publish(result);
    }

}

Il est maintenant prêt à remplir en ScheduledExecutorService:

RScheduledExecutorService executorService = redisson.getExecutorService("myExecutor");
ScheduledFuture<?> future = executorService.schedule(new CallableTask(), 10, 20, TimeUnit.MINUTES);

future.get();
//or cancel it
future.cancel(true);

Exemples avec cron expressions:

executorService.schedule(new RunnableTask(), CronSchedule.of("10 0/5 * * * ?"));

executorService.schedule(new RunnableTask(), CronSchedule.dailyAtHourAndMinute(10, 5));

executorService.schedule(new RunnableTask(), CronSchedule.weeklyOnDayAndHourAndMinute(12, 4, Calendar.MONDAY, Calendar.FRIDAY));

Toutes les tâches sont en cours d'exécution sur Redisson nœud.

0

Une approche combinée semble plausible:
1. Aucune nouvelle tâche d'horodatage peut être moins de temps courant (borne moins). En supposant fiable NTP synchronisation.
2. Toutes les tâches de seau-listes sur les touches, suffixé à la tâche d'horodatage.
3. En outre, toutes les tâches horodateurs aller à la zset (clé et le score — timestamp lui-même).
4. De nouvelles tâches sont acceptés à partir de clients par l'intermédiaire de séparer Redis liste.
5. Boucle: aller Chercher le plus ancien N expiré horodateurs via zrangebyscore ... la limite.
6. BLPOP avec délai sur la nouvelle liste des tâches et des listes extraites des horodateurs.
7. Si j'ai sorti un vieux de la tâche, de la traiter. Si la nouvelle — ajouter dans le seau et zset.
8. Vérifier si le traitement des seaux sont vides. Si oui — supprimer la liste et entrt de zset. Sans doute ne pas vérifier très récemment expiré seaux, pour se protéger contre le temps, les problèmes de synchronisation. Fin de la boucle.
Critique? Des commentaires? Alternatives?

Vous devez vous connecter pour publier un commentaire.