Lors de la mise en œuvre du back-end d'applications Web et d'applications mobiles, même les plus simples, il est devenu courant d'utiliser des outils tels que: bases de données, serveur de messagerie (smtp), serveur redis. L'ensemble des outils utilisés est en constante expansion. Par exemple, les files d'attente de messages, à en juger par le nombre d'installations du package
amqplib (650 000 installations par semaine), sont utilisées avec les bases de données relationnelles (package mysql 460 000 installations par semaine et pg 800 000 installations par semaine).
Aujourd'hui, je veux parler des files d'attente de travaux, qui sont jusqu'à présent utilisées un ordre de grandeur de moins, bien que le besoin s'en fasse sentir, dans presque tous les projets réels.
Ainsi, les files d'attente de travaux vous permettent d'effectuer certaines tâches de manière asynchrone, en fait, d'effectuer une fonction avec les paramètres d'entrée donnés et à l'heure définie.
Selon les paramètres, la tâche peut être effectuée:
- immédiatement après l'ajout à la file d'attente des travaux;
- une fois à une heure définie;
- plusieurs fois dans les délais.
Les files d'attente de travaux vous permettent de transférer des paramètres vers un travail en cours d'exécution, de suivre et de réexécuter les travaux qui ont échoué et de définir une limite sur le nombre de travaux qui s'exécutent simultanément.
La grande majorité des applications sur Node.js sont associées au développement d'une API REST pour les applications Web et mobiles. La réduction du temps d'exécution de l'API REST est importante pour un travail confortable de l'utilisateur avec l'application. Dans le même temps, un appel à l'API REST peut lancer des opérations longues et / ou gourmandes en ressources. Par exemple, après avoir effectué un achat, vous devez envoyer à l'utilisateur un message push vers l'application mobile ou envoyer une demande pour effectuer un achat sur l'API REST CRM. Ces requêtes peuvent être effectuées de manière asynchrone. Comment le faire correctement si vous ne disposez pas d'un outil pour travailler avec les files d'attente de travaux? Par exemple, vous pouvez envoyer un message à la file d'attente de messages, démarrer un travailleur qui lira ces messages et effectuera le travail nécessaire en fonction de ces messages.
En fait, c'est ce que font les files d'attente de travaux. Cependant, si vous regardez attentivement, les files d'attente de travaux présentent plusieurs différences fondamentales par rapport à la file d'attente de messages. Premièrement, les messages (statiques) sont placés dans la file d'attente de messages et les files d'attente de travaux impliquent une sorte de travail (appel de fonction). Deuxièmement, la file d'attente des travaux implique la présence d'un processeur (travailleur) qui effectuera le travail donné. Dans ce cas, des fonctionnalités supplémentaires sont nécessaires. Le nombre de processeurs doit être mis à l'échelle de manière transparente en cas de charge accrue. D'un autre côté, il est nécessaire de limiter le nombre de tâches exécutées simultanément sur un processeur-travailleur afin de lisser les charges de pointe et d'empêcher le déni de service. Cela montre qu'il existe un besoin d'un outil qui pourrait exécuter des tâches asynchrones en définissant divers paramètres, aussi simple que de faire une demande en utilisant l'API REST (ou mieux si c'est encore plus facile).
À l'aide de files d'attente de messages, il est relativement simple d'implémenter une file d'attente de travaux qui s'exécute immédiatement après la mise en file d'attente d'un travail. Mais souvent, il est nécessaire de terminer la tâche une fois à une heure définie ou selon un calendrier. Pour ces tâches, un certain nombre de packages sont largement utilisés qui implémentent la logique cron sous Linux. Afin de ne pas être infondé, je dirai que le package node-cron a 480 000 installations par semaine, node-schedule - 170 000 installations par semaine.
L'utilisation de node-cron est, bien sûr, plus pratique que l'ascétique setInterval (), mais personnellement, j'ai rencontré un certain nombre de problèmes lors de son utilisation. Si pour exprimer un inconvénient général, c'est le manque de contrôle sur le nombre de tâches exécutées simultanément (cela stimule les pics de charge: l'augmentation de la charge ralentit le travail des tâches, le ralentissement des tâches augmente le nombre de tâches exécutées simultanément, ce qui à son tour charge encore plus le système), l'incapacité d'exécuter le nœud pour augmenter la productivité -cron sur plusieurs cœurs (dans ce cas, toutes les tâches sont exécutées indépendamment sur chaque cœur) et le manque d'outils pour suivre et redémarrer les tâches terminées Xia avec une erreur.
J'espère avoir montré que la nécessité d'un tel outil comme la file d'attente des travaux est comparable à des outils tels que les bases de données. Et ces fonds sont apparus, bien qu'ils ne soient pas encore largement utilisés. Je vais énumérer les plus populaires d'entre eux:
Aujourd'hui, je vais considérer l'utilisation du package bull, avec lequel je travaille moi-même. Pourquoi ai-je choisi ce forfait particulier (bien que je n'impose pas mon choix aux autres). À ce moment, lorsque j'ai commencé à chercher une implémentation pratique de la file d'attente de messages, le projet de file d'attente d'abeilles était déjà arrêté. L'implémentation de kue, selon les repères donnés dans le référentiel de la file d'attente d'abeilles, était loin derrière les autres implémentations et, en outre, ne contenait pas les moyens d'exécuter des tâches exécutées périodiquement. Le projet d'agenda implémente des files d'attente avec stockage dans la base de données mongodb. C'est un gros plus pour certains cas, si vous avez besoin d'une super-fiabilité lors du placement de tâches dans la file d'attente. Mais ce n'est pas seulement un facteur décisif. Naturellement, j'ai testé toutes les options d'endurance de la bibliothèque, générant un grand nombre de tâches dans la file d'attente, et je n'ai toujours pas pu obtenir un travail ininterrompu de l'agenda. En dépassant un certain nombre de tâches, l'agenda s'est arrêté et a cessé de mettre des tâches à exécution.
Par conséquent, j'ai opté pour bull qui implémente une API pratique, avec une vitesse et une évolutivité suffisantes, car le package bull utilise un serveur redis comme backend. En particulier, vous pouvez utiliser un cluster de serveurs redis.
Lors de la création d'une file d'attente, il est très important de sélectionner les paramètres optimaux pour la file d'attente des travaux. Il existe de nombreux paramètres, et la valeur de certains d'entre eux ne m'a pas atteint tout de suite. Après de nombreuses expériences, je me suis installé sur les paramètres suivants:
const Bull = require('bull'); const redis = { host: 'localhost', port: 6379, maxRetriesPerRequest: null, connectTimeout: 180000 }; const defaultJobOptions = { removeOnComplete: true, removeOnFail: false, }; const limiter = { max: 10000, duration: 1000, bounceBack: false, }; const settings = { lockDuration: 600000,
Dans des cas triviaux, il n'est pas nécessaire de créer de nombreuses files d'attente, car dans chaque file d'attente, vous pouvez spécifier des noms pour différentes tâches et associer un processeur-travailleur à chaque nom:
const { bull } = require('../bull'); bull.process('push:news', 1, `${__dirname}/push-news.js`); bull.process('push:status', 2, `${__dirname}/push-status.js`); ... bull.process('some:job', function(...args) { ... });
J'utilise l'opportunité qui se présente au taureau «prêt à l'emploi» - pour paralléliser les processeurs sur plusieurs cœurs. Pour ce faire, le deuxième paramètre définit le nombre de cœurs sur lesquels le processeur-travailleur sera lancé, et dans le troisième paramètre, le nom du fichier avec la définition de la fonction de traitement des travaux. Si une telle fonctionnalité n'est pas nécessaire, vous pouvez simplement passer une fonction de rappel comme deuxième paramètre.
La tâche est mise en file d'attente par un appel à la méthode add (), à laquelle le nom et l'objet de la file d'attente sont passés dans les paramètres, qui seront ensuite transmis au gestionnaire de tâches. Par exemple, dans un hook ORM, après avoir créé une entrée avec de nouvelles actualités, je peux envoyer un message push de manière asynchrone à tous les clients:
afterCreate(instance) { bull.add('push:news', _.pick(instance, 'id', 'title', 'message'), options); }
Le gestionnaire d'événements accepte dans les paramètres l'objet de tâche avec les paramètres passés à la méthode add () et la fonction done (), qui doivent être appelées pour confirmer que la tâche est terminée ou pour informer que la tâche s'est terminée avec une erreur:
const { firebase: { admin } } = require('../firebase'); const { makePayload } = require('./makePayload'); module.exports = (job, done) => { const { id, title, message } = job.data; const data = { id: String(id), type: 'news', }; const payloadRu = makePayload(title.ru, message.ru, data); const payloadEn = makePayload(title.en, message.en, data); return Promise.all([ admin.messaging().send({ ...payloadRu, condition: "'news' in topics && 'ru' in topics" }), admin.messaging().send({ ...payloadEn, condition: "'news' in topics && 'en' in topics" }), ]) .then(response => done(null, response)) .catch(done); };
Pour afficher l'état de la file d'attente des travaux, vous pouvez utiliser l'outil arena-bull:
const Arena = require('bull-arena'); const redis = { host: 'localhost', port: 6379, maxRetriesPerRequest: null, connectTimeout: 180000 }; const arena = Arena({ queues: [ { name: 'my_gueue', hostId: 'My Queue', redis, }, ], }, { basePath: '/', disableListen: true, }); module.exports = { arena };
Et enfin, un petit hack de vie. Comme je l'ai dit, bull utilise un serveur redis comme backend. Lorsque le serveur redis est redémarré, la probabilité de disparition du travail est très faible. Mais sachant que les administrateurs système peuvent parfois simplement «vider le cache de radis», tout en supprimant toutes les tâches en particulier, j'étais principalement préoccupé par l'exécution périodique de tâches, qui dans ce cas s'arrêtaient pour toujours. À cet égard, j'ai trouvé l'occasion de reprendre ces tâches périodiques:
const cron = '*/10 * * * * *'; const { bull } = require('./app/services/bull'); bull.getRepeatableJobs() .then(jobs => Promise.all(_.map(jobs, (job) => { const [name, cron] = job.key.split(/:{2,}/); return bull.removeRepeatable(name, { cron }); }))) .then(() => bull.add('check:status', {}, { priority: 1, repeat: { cron } })); setInterval(() => bull.add('check:status', {}, { priority: 1, repeat: { cron } }), 60000);
Autrement dit, la tâche est d'abord exclue de la file d'attente, puis définie à nouveau, et tout cela (hélas) par setInterval (). En fait, sans un tel piratage de la vie, je n'aurais probablement pas décidé d'utiliser des tâches périodiques sur le taureau.
apapacy@gmail.com
3 juillet 2019