RabbitMQ / AMQP: file d'attente unique, plusieurs consommateurs pour le même message?

Je commence tout juste à utiliser RabbitMQ et AMQP en général.

• J'ai une file de messages
• J'ai plusieurs consommateurs, ce que j'aimerais faire différemment avec le même message .

La plupart de la documentation de RabbitMQ semble se concentrer sur le round-robin, c'est-à-dire où un seul message est consommé par un seul consommateur, la charge étant répartie entre chaque consommateur. C'est en effet le comportement dont je suis témoin.

Un exemple: le producteur a une seule file d'attente et envoie des messages toutes les 2 s:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;

connection.on('ready', function () {
  var sendMessage = function(connection, queue_name, payload) {
    var encoded_payload = JSON.stringify(payload);  
    connection.publish(queue_name, encoded_payload);
  }

  setInterval( function() {    
    var test_message = 'TEST '+count
    sendMessage(connection, "my_queue_name", test_message)  
    count += 1;
  }, 2000) 


})

Et voici un consommateur:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
connection.on('ready', function () {
  connection.queue("my_queue_name", function(queue){
    queue.bind('#'); 
    queue.subscribe(function (message) {
      var encoded_payload = unescape(message.data)
      var payload = JSON.parse(encoded_payload)
      console.log('Recieved a message:')
      console.log(payload)
    })
  })
})

Si je lance le consommateur deux fois, je peux voir que chaque consommateur consomme des messages alternatifs dans un comportement circulaire. Par exemple, je vais voir les messages 1, 3, 5 dans un terminal, 2, 4, 6 dans l'autre .

Ma question est:

• Puis-je faire en sorte que chaque consommateur reçoive les mêmes messages? Ie, les deux consommateurs reçoivent le message 1, 2, 3, 4, 5, 6? Comment cela s'appelle-t-il dans AMQP / RabbitMQ parler? Comment est-il normalement configuré?

• Cela se fait-il couramment? Dois-je simplement demander à l'échange de router le message dans deux files d'attente distinctes, avec un seul consommateur?

Puis-je faire en sorte que chaque consommateur reçoive les mêmes messages? Ie, les deux consommateurs reçoivent le message 1, 2, 3, 4, 5, 6? Comment cela s'appelle-t-il dans AMQP / RabbitMQ parler? Comment est-il normalement configuré?

Non, pas si les consommateurs sont sur la même file d'attente. À partir du guide AMQP Concepts de RabbitMQ :

il est important de comprendre que, dans AMQP 0-9-1, la charge des messages est équilibrée entre les consommateurs.

Cela semble impliquer que le comportement à tour de rôle dans une file d'attente est un fait donné et non configurable. Par exemple, des files d'attente séparées sont nécessaires pour que le même ID de message soit géré par plusieurs consommateurs.

Cela se fait-il couramment? Dois-je simplement demander à l'échange de router le message dans deux files d'attente distinctes, avec un seul consommateur?

Non, ce n'est pas le cas, une seule file d'attente / plusieurs consommateurs avec chaque consommateur gérant le même ID de message n'est pas possible. Avoir l'échange acheminer le message dans deux files d'attente séparées est en effet mieux.

Comme je n'ai pas besoin d'un routage trop complexe, un échange de fanout gérera cela bien. Je ne me suis pas trop concentré sur les échanges plus tôt car node-amqp a le concept d'un `` échange par défaut '' vous permettant de publier des messages directement sur une connexion, mais la plupart des messages AMQP sont publiés dans un échange spécifique.

Voici mon échange de fanout, à la fois d'envoi et de réception:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;

connection.on('ready', function () {
  connection.exchange("my_exchange", options={type:'fanout'}, function(exchange) {   

    var sendMessage = function(exchange, payload) {
      console.log('about to publish')
      var encoded_payload = JSON.stringify(payload);
      exchange.publish('', encoded_payload, {})
    }

    // Recieve messages
    connection.queue("my_queue_name", function(queue){
      console.log('Created queue')
      queue.bind(exchange, ''); 
      queue.subscribe(function (message) {
        console.log('subscribed to queue')
        var encoded_payload = unescape(message.data)
        var payload = JSON.parse(encoded_payload)
        console.log('Recieved a message:')
        console.log(payload)
      })
    })

    setInterval( function() {    
      var test_message = 'TEST '+count
      sendMessage(exchange, test_message)  
      count += 1;
    }, 2000) 
 })
})

Lisez simplement le tutoriel rabbitmq . Vous publiez le message pour échanger, pas pour mettre en file d'attente; il est ensuite acheminé vers les files d'attente appropriées. Dans votre cas, vous devez lier une file d'attente distincte pour chaque consommateur. De cette façon, ils peuvent consommer des messages de manière totalement indépendante.

Les deux dernières réponses sont presque correctes - j'ai des tonnes d'applications qui génèrent des messages qui doivent aboutir avec différents consommateurs, donc le processus est très simple.

Si vous souhaitez que plusieurs consommateurs reçoivent le même message, procédez comme suit.

Créez plusieurs files d'attente, une pour chaque application qui doit recevoir le message, dans chaque propriété de file d'attente, «liez» une balise de routage avec l'échange amq.direct. Changez votre application de publication pour l'envoyer à amq.direct et utilisez la balise de routage (pas une file d'attente). AMQP copiera ensuite le message dans chaque file d'attente avec la même liaison. Fonctionne comme un charme :)

Exemple: Disons que j'ai une chaîne JSON que je génère, je la publie sur l'échange "amq.direct" en utilisant la balise de routage "new-sales-order", j'ai une file d'attente pour mon application order_printer qui imprime la commande, j'ai un file d'attente pour mon système de facturation qui enverra une copie de la commande et facturera le client et j'ai un système d'archives Web dans lequel j'archive les commandes pour des raisons historiques / de conformité et j'ai une interface Web client où les commandes sont suivies au fur et à mesure que d'autres informations arrivent une commande.

Donc mes files d'attente sont: order_printer, order_billing, order_archive et order_tracking Tous ont la balise de liaison "new-sales-order" qui leur est liée, les 4 recevront les données JSON.

C'est un moyen idéal d'envoyer des données sans que l'application de publication ne sache ou ne se soucie des applications de réception.

Oui, chaque consommateur peut recevoir les mêmes messages. jetez un œil à http://www.rabbitmq.com/tutorials/tutorial-three-python.html http://www.rabbitmq.com/tutorials/tutorial-four-python.html http: //www.rabbitmq. com / tutoriels / tutoriel-cinq-python.html

pour différentes manières d'acheminer les messages. Je sais qu'ils sont pour python et java, mais il est bon de comprendre les principes, de décider de ce que vous faites et ensuite de trouver comment le faire dans JS. On dirait que vous voulez faire un simple déploiement ( tutoriel 3 ), qui envoie les messages à toutes les files d'attente connectées à l'échange.

La différence avec ce que vous faites et ce que vous voulez faire est essentiellement que vous allez configurer et échanger ou taper fanout. Les excahnges de fanout envoient tous les messages à toutes les files d'attente connectées. Chaque file d'attente aura un consommateur qui aura accès à tous les messages séparément.

Oui, cela se fait couramment, c'est l'une des fonctionnalités de l'AMPQ.

Le modèle d'envoi est une relation un-à-un. Si vous voulez "envoyer" à plus d'un récepteur, vous devez utiliser le modèle pub / sub. Voir http://www.rabbitmq.com/tutorials/tutorial-three-python.html pour plus de détails.

RabbitMQ / AMQP: file d'attente unique, plusieurs consommateurs pour le même message et l'actualisation de la page.

rabbit.on('ready', function () {    });
    sockjs_chat.on('connection', function (conn) {

        conn.on('data', function (message) {
            try {
                var obj = JSON.parse(message.replace(/\r/g, '').replace(/\n/g, ''));

                if (obj.header == "register") {

                    // Connect to RabbitMQ
                    try {
                        conn.exchange = rabbit.exchange(exchange, { type: 'topic',
                            autoDelete: false,
                            durable: false,
                            exclusive: false,
                            confirm: true
                        });

                        conn.q = rabbit.queue('my-queue-'+obj.agentID, {
                            durable: false,
                            autoDelete: false,
                            exclusive: false
                        }, function () {
                            conn.channel = 'my-queue-'+obj.agentID;
                            conn.q.bind(conn.exchange, conn.channel);

                            conn.q.subscribe(function (message) {
                                console.log("[MSG] ---> " + JSON.stringify(message));
                                conn.write(JSON.stringify(message) + "\n");
                            }).addCallback(function(ok) {
                                ctag[conn.channel] = ok.consumerTag; });
                        });
                    } catch (err) {
                        console.log("Could not create connection to RabbitMQ. \nStack trace -->" + err.stack);
                    }

                } else if (obj.header == "typing") {

                    var reply = {
                        type: 'chatMsg',
                        msg: utils.escp(obj.msga),
                        visitorNick: obj.channel,
                        customField1: '',
                        time: utils.getDateTime(),
                        channel: obj.channel
                    };

                    conn.exchange.publish('my-queue-'+obj.agentID, reply);
                }

            } catch (err) {
                console.log("ERROR ----> " + err.stack);
            }
        });

        // When the visitor closes or reloads a page we need to unbind from RabbitMQ?
        conn.on('close', function () {
            try {

                // Close the socket
                conn.close();

                // Close RabbitMQ           
               conn.q.unsubscribe(ctag[conn.channel]);

            } catch (er) {
                console.log(":::::::: EXCEPTION SOCKJS (ON-CLOSE) ::::::::>>>>>>> " + er.stack);
            }
        });
    });

Pour obtenir le comportement souhaité, demandez simplement à chaque consommateur de consommer à partir de sa propre file d'attente. Vous devrez utiliser un type d'échange non direct (sujet, en-tête, fanout) afin de transmettre le message à toutes les files d'attente à la fois.

Comme j'évalue votre cas, c'est:

• J'ai une file d'attente de messages (votre source pour recevoir les messages, appelons-la q111)

• J'ai plusieurs consommateurs, ce que j'aimerais faire différemment avec le même message.

Votre problème ici est que lorsque 3 messages sont reçus par cette file d'attente, le message 1 est consommé par un consommateur A, les autres consommateurs B et C consomment les messages 2 et 3. Là où vous avez besoin d'une configuration où rabbitmq transmet les mêmes copies de tous ces trois messages (1, 2, 3) aux trois consommateurs connectés (A, B, C) simultanément.

Bien que de nombreuses configurations puissent être faites pour y parvenir, un moyen simple consiste à utiliser le concept en deux étapes suivant:

• Utilisez une pelle rabbitmq dynamique pour récupérer les messages de la file d'attente souhaitée (q111) et les publier dans un échange de fanout (échange exclusivement créé et dédié à cet effet).
• Maintenant, reconfigurez vos consommateurs A, B et C (qui écoutaient la file d'attente (q111)) pour écouter directement depuis cet échange Fanout en utilisant une file d'attente exclusive et anonyme pour chaque consommateur.

Remarque: lorsque vous utilisez ce concept, ne consommez pas directement à partir de la file d'attente source (q111), car les messages déjà consommés ne seront pas transférés vers votre échange Fanout.

Si vous pensez que cela ne répond pas exactement à vos exigences ... n'hésitez pas à poster vos suggestions :-)

Si vous utilisez la bibliothèque amqplib comme moi, ils ont un exemple pratique d'implémentation du tutoriel Publish / Subscribe RabbitMQ que vous pourriez trouver utile.

Je pense que vous devriez vérifier l'envoi de vos messages en utilisant le fan-out échangeur. De cette façon, vous recevrez le même message pour différents consommateurs, sous le tableau RabbitMQ crée des files d'attente différentes pour chacun de ces nouveaux consommateurs / abonnés.

C'est le lien pour voir l'exemple du tutoriel en javascript https://www.rabbitmq.com/tutorials/tutorial-one-javascript.html

Il y a une option intéressante dans ce scénario que je n'ai pas trouvée dans les réponses ici.

Vous pouvez Nack des messages avec la fonction de "mise en file d'attente" dans un consommateur pour les traiter dans un autre. En règle générale, ce n'est pas une bonne façon, mais peut-être que ce sera assez bon pour quelqu'un.

https://www.rabbitmq.com/nack.html

Et méfiez-vous des boucles (quand tous les concumers nack + message en file d'attente)!