RabbitMQ / AMQP: antrian tunggal, banyak konsumen untuk pesan yang sama?

Saya baru mulai menggunakan RabbitMQ dan AMQP secara umum.

• Saya memiliki antrian pesan
• Saya memiliki banyak konsumen, yang ingin saya lakukan hal yang berbeda dengan pesan yang sama .

Sebagian besar dokumentasi RabbitMQ tampaknya difokuskan pada round-robin, yaitu di mana satu pesan dikonsumsi oleh satu konsumen, dengan beban yang tersebar di antara masing-masing konsumen. Ini memang perilaku yang saya saksikan.

Contoh: produser memiliki antrian tunggal, dan mengirim pesan setiap 2 detik:

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) 


})

Dan inilah konsumennya:

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)
    })
  })
})

Jika saya memulai konsumen dua kali, saya dapat melihat bahwa setiap konsumen mengkonsumsi pesan alternatif dalam perilaku round-robin. Misalnya, saya akan melihat pesan 1, 3, 5 di satu terminal, 2, 4, 6 di terminal lain .

Pertanyaanku adalah:

• Bisakah saya meminta setiap konsumen menerima pesan yang sama? Yaitu, kedua konsumen mendapatkan pesan 1, 2, 3, 4, 5, 6? Apakah ini yang disebut dalam pembicaraan AMQP / RabbitMQ? Bagaimana biasanya dikonfigurasi?

• Apakah ini biasa dilakukan? Haruskah saya meminta pertukaran rute pesan menjadi dua antrian terpisah, dengan satu konsumen, sebagai gantinya?

Bisakah saya meminta setiap konsumen menerima pesan yang sama? Yaitu, kedua konsumen mendapatkan pesan 1, 2, 3, 4, 5, 6? Apakah ini yang disebut dalam pembicaraan AMQP / RabbitMQ? Bagaimana biasanya dikonfigurasi?

Tidak, tidak jika konsumen berada pada antrian yang sama. Dari panduan Konsep AMQP RabbitMQ :

penting untuk dipahami bahwa, dalam AMQP 0-9-1, pesan dimuat secara seimbang di antara konsumen.

Ini tampaknya menyiratkan bahwa perilaku round-robin dalam antrian adalah sesuatu yang diberikan , dan tidak dapat dikonfigurasi. Yaitu, diperlukan antrian terpisah agar ID pesan yang sama ditangani oleh banyak konsumen.

Apakah ini biasa dilakukan? Haruskah saya meminta pertukaran rute pesan menjadi dua antrian terpisah, dengan satu konsumen, sebagai gantinya?

Tidak, antrian tunggal / banyak konsumen dengan masing-masing konsumen menangani ID pesan yang sama tidak mungkin. Memiliki pertukaran rute pesan ke dalam dua antrian terpisah memang lebih baik.

Karena saya tidak memerlukan perutean yang terlalu rumit, pertukaran fanout akan menangani ini dengan baik. Saya tidak terlalu fokus pada Pertukaran sebelumnya karena node-amqp memiliki konsep 'pertukaran default' yang memungkinkan Anda untuk mempublikasikan pesan ke koneksi secara langsung, namun sebagian besar pesan AMQP diterbitkan untuk pertukaran tertentu.

Inilah pertukaran penggemar saya, baik mengirim dan menerima:

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) 
 })
})

Baca saja tutorial rabbitmq . Anda mempublikasikan pesan untuk dipertukarkan, bukan untuk mengantri; kemudian dialihkan ke antrian yang sesuai. Dalam kasus Anda, Anda harus mengikat antrian terpisah untuk setiap konsumen. Dengan begitu, mereka dapat menggunakan pesan sepenuhnya secara mandiri.

Beberapa jawaban terakhir hampir benar - Saya punya banyak aplikasi yang menghasilkan pesan yang harus berakhir dengan konsumen yang berbeda sehingga prosesnya sangat sederhana.

Jika Anda ingin banyak konsumen untuk pesan yang sama, lakukan prosedur berikut.

Buat beberapa antrian, satu untuk setiap aplikasi yang menerima pesan, di setiap properti antrian, "ikat" tag perutean dengan pertukaran amq.direct. Ubah aplikasi penerbitan Anda untuk mengirim ke amq.direct dan menggunakan tag-rute (bukan antrian). AMQP kemudian akan menyalin pesan ke setiap antrian dengan ikatan yang sama. Bekerja seperti pesona :)

Contoh: Katakanlah saya punya string JSON yang saya hasilkan, saya publikasikan ke pertukaran "amq.direct" menggunakan tag perutean "new-sales-order", saya punya antrean untuk aplikasi order_printer saya yang mencetak pesanan, saya punya mengantri untuk sistem penagihan saya yang akan mengirim salinan pesanan dan menagih klien dan saya memiliki sistem arsip web di mana saya mengarsipkan pesanan untuk alasan historis / kepatuhan dan saya memiliki antarmuka web klien di mana pesanan dilacak saat info lain masuk tentang sebuah pesanan.

Jadi antrian saya adalah: order_printer, order_billing, order_archive dan order_tracking Semua memiliki tag pengikat "new-sales-order" terikat dengan mereka, semua 4 akan mendapatkan data JSON.

Ini adalah cara yang ideal untuk mengirim data tanpa mengetahui atau memperhatikan aplikasi aplikasi penerima.

Ya, setiap konsumen dapat menerima pesan yang sama. lihat di http://www.rabbitmq.com/tutorials/tutorial-three-python.html http://www.rabbitmq.com/tutorials/tutorial-four-python.html http: //www.rabbitmq. com / tutorials / tutorial-five-python.html

untuk berbagai cara merutekan pesan. Saya tahu mereka untuk python dan java tetapi bagus untuk memahami prinsip-prinsip, memutuskan apa yang Anda lakukan dan kemudian menemukan cara melakukannya di JS. Kedengarannya seperti Anda ingin melakukan fanout sederhana ( tutorial 3 ), yang mengirim pesan ke semua antrian yang terhubung ke pertukaran.

Perbedaan dengan apa yang Anda lakukan dan apa yang ingin Anda lakukan pada dasarnya adalah bahwa Anda akan mengatur dan bertukar atau mengetik fanout. Fanout fanah mengirim semua pesan ke semua antrian yang terhubung. Setiap antrian akan memiliki konsumen yang akan memiliki akses ke semua pesan secara terpisah.

Ya ini biasa dilakukan, itu adalah salah satu fitur AMPQ.

Pola kirim adalah hubungan satu-ke-satu. Jika Anda ingin "mengirim" ke lebih dari satu penerima, Anda harus menggunakan pola pub / sub. Lihat http://www.rabbitmq.com/tutorials/tutorial-three-python.html untuk lebih jelasnya.

RabbitMQ / AMQP: antrian tunggal, banyak konsumen untuk pesan dan penyegaran halaman yang sama.

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);
            }
        });
    });

Untuk mendapatkan perilaku yang Anda inginkan, minta setiap konsumen untuk mengkonsumsi dari antriannya sendiri. Anda harus menggunakan jenis pertukaran tidak langsung (topik, header, fanout) untuk mendapatkan pesan ke semua antrian sekaligus.

Saat saya menilai kasus Anda adalah:

• Saya memiliki antrian pesan (sumber Anda untuk menerima pesan, beri nama q111)

• Saya memiliki banyak konsumen, yang ingin saya lakukan hal yang berbeda dengan pesan yang sama.

Masalah Anda di sini adalah ketika 3 pesan diterima oleh antrian ini, pesan 1 dikonsumsi oleh konsumen A, konsumen lain B dan C mengkonsumsi pesan 2 dan 3. Ketika Anda membutuhkan pengaturan di mana rabbitmq meneruskan salinan yang sama dari ketiga pesan ini (1,2,3) untuk ketiga konsumen yang terhubung (A, B, C) secara bersamaan.

Sementara banyak konfigurasi dapat dibuat untuk mencapai ini, cara sederhana adalah dengan menggunakan konsep dua langkah berikut:

• Gunakan shovel rabbitmq dinamis untuk mengambil pesan dari antrian yang diinginkan (q111) dan mempublikasikannya ke pertukaran fanout (pertukaran yang khusus dibuat dan didedikasikan untuk tujuan ini).
• Sekarang konfigurasikan ulang konsumen Anda A, B & C (yang mendengarkan antrian (q111)) untuk mendengarkan dari pertukaran Fanout ini secara langsung menggunakan antrian eksklusif & anonim untuk setiap konsumen.

Catatan: Saat menggunakan konsep ini jangan mengkonsumsi langsung dari antrian sumber (q111), karena pesan yang sudah dikonsumsi tidak akan disekop ke pertukaran Fanout Anda.

Jika menurut Anda ini tidak memenuhi persyaratan persis Anda ... jangan ragu untuk mengirim saran Anda :-)

Jika Anda kebetulan menggunakan pustaka amqplib seperti saya, mereka memiliki contoh praktis dari implementasi tutorial Publikasikan / Berlangganan RabbitMQ yang mungkin berguna bagi Anda.

Saya pikir Anda harus memeriksa pengiriman pesan Anda menggunakan exchanger fan-out . Dengan begitu Anda akan menerima pesan yang sama untuk konsumen yang berbeda, di bawah meja RabbitMQ membuat antrian yang berbeda untuk masing-masing konsumen / pelanggan baru ini.

Ini adalah tautan untuk melihat contoh tutorial di javascript https://www.rabbitmq.com/tutorials/tutorial-one-javascript.html

Ada satu opsi menarik dalam skenario ini yang belum saya temukan dalam jawaban di sini.

Anda dapat Nack pesan dengan fitur "requeue" di satu konsumen untuk memprosesnya di yang lain. Secara umum itu bukan cara yang benar, tetapi mungkin itu akan cukup baik untuk seseorang.

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

Dan waspadalah terhadap loop (ketika semua konsumen nack + pesan requeue)!