Les files d'attente de messages sont utilisées pour effectuer: opérations en attente, interaction entre services, «traitement par lots», etc. Il existe des solutions spécialisées pour organiser de telles files d'attente, telles que: RabbitMQ, ActiveMQ, ZeroMQ, etc., mais il arrive souvent qu'elles ne soient pas d'un grand besoin, et leur installation et leur support causeront plus de douleur et de souffrance qu'elles n'en apporteront d'avantages. Supposons que vous ayez un service lors de votre inscription dans lequel un e-mail est envoyé à l'utilisateur pour confirmation, et si vous utilisez Postgres, alors vous avez de la chance - dans Postgres, presque prêt à l'emploi, il existe une extension PgQ qui fera tout le sale boulot pour vous.
Dans cet article, je parlerai de la mise en file d'attente des messages (tâches) dans PostgreSQL en utilisant l'extension PgQ. Cet article sera utile si vous n'avez pas utilisé PgQ ou utilisez des files d'attente auto-écrites au-dessus de Postgres.
Pourquoi avez-vous besoin de PgQ, si vous pouvez simplement créer une tablette et y écrire des tâches? Cela semble possible, mais vous devrez prendre en compte l'accès parallèle aux tâches, les erreurs possibles (que se passera-t-il si le processus de traitement de la tâche tombe?), Ainsi que les performances (PgQ est très rapide, et les solutions auto-écrites ne le sont généralement pas, surtout si la transaction est en la base de données ne se ferme pas pendant toute l'exécution de la tâche), mais la raison la plus importante à mon avis pour laquelle il est nécessaire d'utiliser PgQ est que PgQ est déjà écrit et fonctionne, et la solution auto-écrite doit encore être écrite (UPD: pourquoi cela ne vaut pas la peine d'utiliser des files d'attente auto-écrites , vous pouvez lire, par exemple,
ici ).
(UPD: puisque PgQ s'exécute au-dessus de Postgres, tous les plaisirs des transactions peuvent également y être utilisés)
Mais PgQ a un énorme inconvénient - le manque de documentation, j'essaie de compenser cette lacune avec cet article.
Périphérique
PgQ se compose de parties (au moins 2): 1 - extension pgq pour postgres, 2 - démon pgqd (pour les installer un peu plus tard).
Toutes les interactions avec la file d'attente sont effectuées à l'aide de fonctions à l'intérieur de Postgres.
Par exemple, pour créer une file d'attente, vous devez exécuter
select * from pgq.create_queue({ } text);
Une fois la file d'attente créée, vous pouvez y ajouter des messages.
select * from pgq.insert_event({ } text, { } text, { } text);
Maintenant, nous devons apprendre à recevoir des messages enregistrés. Pour cela, il existe une entité telle que «consommateur» (j'écrirai un consommateur), qui reçoit non pas des messages (événements), mais «batch» (batch). Un bach est un groupe de messages consécutifs, les baches sont créés à l'aide de pgqd. Périodiquement (le paramètre «ticker_period» dans le fichier de configuration) pgqd prend tous les messages accumulés et écrit dans un nouveau bach.
Il est important que pgqd ne fonctionne pas, alors de nouveaux bachs ne sont pas créés, ce qui signifie que les consommateurs n'ont rien à lire, également si pgqd n'a pas fonctionné pendant une longue période puis allumé, il créera un gros bach à partir des messages accumulés pendant ce temps, donc pgqd ne devrait pas être il suffit de l'éteindre.
Enregistrement d'un consommateur (
Important! Un consommateur ne recevra des événements enregistrés
qu'après son enregistrement, vous devez donc d'abord créer un consommateur, puis écrire des événements uniquement):
select * from pgq.register_consumer({ } text, { } text);
(similaire à pgq.unregister_consumer)Chaque consommateur recevra
absolument tous les événements survenus après sa création (même traités par un autre consommateur), ce qui signifie que vous n'avez probablement besoin que d'un seul consommateur pour un tour. Ensuite, je vais vous dire comment répartir la charge sur plusieurs serveurs.
Pour obtenir un bach, vous devez d'abord trouver son ID:
select * from pgq.next_batch({ } text, { } text);
La fonction peut retourner NULL si le consommateur a traité tous les bachs. Dans ce cas, il vous suffit d'attendre que pgqd crée un nouveau bach.
Dans ce cas, alors que le bach n'est pas traité, cette fonction retournera la même valeur.
Vous pouvez obtenir tous les événements du lot en utilisant:
select * from pgq.get_batch_events({id } bigint);
(Le bach est peut-être vide.)
Si une erreur s'est produite lors du traitement de l'un d'eux, vous pouvez essayer de traiter cet événement ultérieurement:
select * from pgq.event_retry({id } bigint, {id } bigint, { } integer);
Pour informer sur la fin du bach et avoir l'opportunité d'en démarrer un nouveau, il est utilisé
select * from pgq.finish_batch({id } bigint);
Bien sûr, ce ne sont pas toutes les fonctions de l'extension, je recommande de lire
pgq.imtqy.com/extension/pgq/files/external-sql.html et
github.com/pgq/pgq/tree/master/functions (chaque fichier contient une définition et une description fonction correspondante).
Partage de charge
Afin de gérer les événements simultanément par plusieurs gestionnaires, il existe l'extension pgq_coop, qui fonctionne comme pgq et ajoute une nouvelle entité appelée «sous-consommateur», qui recevra tous les événements à partir du moment de l'enregistrement du consommateur parent, bien sûr, à l'exception de ceux déjà traités.
select * from pgq_coop.register_subconsumer({ } text, { } text, { } text);
select * from pgq_coop.next_batch({ } text, { } text, { } text);
select * from pgq_coop.next_batch({ } text, { } text, { } text, { , } interval);
select * from pgq_coop.finish_batch({id } bigint);
Découvrez toutes les fonctionnalités
ici .
L'installation
L'extension pgq et le démon pgqd sont inclus dans les référentiels PGDG et sont installés très simplement dans la plupart des distributions, par exemple dans Debian:
sudo apt install postgresql-XX-pgq3 pgqd (XX est le numéro de version).
pgqd est un petit programme que vous pouvez découvrir sur l'utilisation de
pgqd --help , n'oubliez pas de l'ajouter à l'
sudo systemctl enable pgqd.service automatique (
sudo systemctl enable pgqd.service , et la configuration par défaut est
/etc/pgqd.ini ).
Pour commencer à utiliser PgQ, dans la base de données il vous suffit de connecter l'extension:
create extension if not exists pgq;
Avec pgq_coop, tout est un peu plus compliqué, ce n'est pas dans le dépôt, mais ce n'est pas difficile de le compiler à partir des sources (exemple pour Debian):
sudo apt install postgresql-server-dev-XX git clone https://github.com/pgq/pgq-coop.git cd pgq-coop sudo make install
et connectez l'extension à l'aide
create extension if not exists pgq_coop;
Liens utiles
Documentation PGQFonctions PgqFonctions Pgq_coopCode source pgqdcompte github avec tous les projets associésPostgres wiki