RabbitMQ - HackMD

owned this note owned this note

<style> .textleft { text-align:left; } .reveal, .reveal h1, .reveal h2, .reveal h3, .reveal h4, .reveal h5, .reveal h6 { font-family:Arial, Microsoft JhengHei;} .small-font { font-size: 20px !important; } .reveal .progress { height: 20px !important; transform: scale(1,2); } .progress span { background: url(data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAAMCAIAAAAs6UAAAAAAGXRFWHRTb2Z0d2FyZQBBZG9iZSBJbWFnZVJlYWR5ccllPAAAAyJpVFh0WE1MOmNvbS5hZG9iZS54bXAAAAAAADw/eHBhY2tldCBiZWdpbj0i77u/IiBpZD0iVzVNME1wQ2VoaUh6cmVTek5UY3prYzlkIj8+IDx4OnhtcG1ldGEgeG1sbnM6eD0iYWRvYmU6bnM6bWV0YS8iIHg6eG1wdGs9IkFkb2JlIFhNUCBDb3JlIDUuMy1jMDExIDY2LjE0NTY2MSwgMjAxMi8wMi8wNi0xNDo1NjoyNyAgICAgICAgIj4gPHJkZjpSREYgeG1sbnM6cmRmPSJodHRwOi8vd3d3LnczLm9yZy8xOTk5LzAyLzIyLXJkZi1zeW50YXgtbnMjIj4gPHJkZjpEZXNjcmlwdGlvbiByZGY6YWJvdXQ9IiIgeG1sbnM6eG1wPSJodHRwOi8vbnMuYWRvYmUuY29tL3hhcC8xLjAvIiB4bWxuczp4bXBNTT0iaHR0cDovL25zLmFkb2JlLmNvbS94YXAvMS4wL21tLyIgeG1sbnM6c3RSZWY9Imh0dHA6Ly9ucy5hZG9iZS5jb20veGFwLzEuMC9zVHlwZS9SZXNvdXJjZVJlZiMiIHhtcDpDcmVhdG9yVG9vbD0iQWRvYmUgUGhvdG9zaG9wIENTNiAoV2luZG93cykiIHhtcE1NOkluc3RhbmNlSUQ9InhtcC5paWQ6QUNCQzIyREQ0QjdEMTFFMzlEMDM4Qzc3MEY0NzdGMDgiIHhtcE1NOkRvY3VtZW50SUQ9InhtcC5kaWQ6QUNCQzIyREU0QjdEMTFFMzlEMDM4Qzc3MEY0NzdGMDgiPiA8eG1wTU06RGVyaXZlZEZyb20gc3RSZWY6aW5zdGFuY2VJRD0ieG1wLmlpZDpBQ0JDMjJEQjRCN0QxMUUzOUQwMzhDNzcwRjQ3N0YwOCIgc3RSZWY6ZG9jdW1lbnRJRD0ieG1wLmRpZDpBQ0JDMjJEQzRCN0QxMUUzOUQwMzhDNzcwRjQ3N0YwOCIvPiA8L3JkZjpEZXNjcmlwdGlvbj4gPC9yZGY6UkRGPiA8L3g6eG1wbWV0YT4gPD94cGFja2V0IGVuZD0iciI/PovDFgYAAAAmSURBVHjaYvjPwMAAxjMZmBhA9H8INv4P4TPM/A+m04zBNECAAQBCWQv9SUQpVgAAAABJRU5ErkJggg==) repeat-x !important; } .progress span:after, .progress span.nyancat { content: ""; background: url(data:image/gif;base64,R0lGODlhIgAVAKIHAL3/9/+Zmf8zmf/MmZmZmf+Z/wAAAAAAACH/C05FVFNDQVBFMi4wAwEAAAAh/wtYTVAgRGF0YVhNUDw/eHBhY2tldCBiZWdpbj0i77u/IiBpZD0iVzVNME1wQ2VoaUh6cmVTek5UY3prYzlkIj8+IDx4OnhtcG1ldGEgeG1sbnM6eD0iYWRvYmU6bnM6bWV0YS8iIHg6eG1wdGs9IkFkb2JlIFhNUCBDb3JlIDUuMy1jMDExIDY2LjE0NTY2MSwgMjAxMi8wMi8wNi0xNDo1NjoyNyAgICAgICAgIj4gPHJkZjpSREYgeG1sbnM6cmRmPSJodHRwOi8vd3d3LnczLm9yZy8xOTk5LzAyLzIyLXJkZi1zeW50YXgtbnMjIj4gPHJkZjpEZXNjcmlwdGlvbiByZGY6YWJvdXQ9IiIgeG1sbnM6eG1wTU09Imh0dHA6Ly9ucy5hZG9iZS5jb20veGFwLzEuMC9tbS8iIHhtbG5zOnN0UmVmPSJodHRwOi8vbnMuYWRvYmUuY29tL3hhcC8xLjAvc1R5cGUvUmVzb3VyY2VSZWYjIiB4bWxuczp4bXA9Imh0dHA6Ly9ucy5hZG9iZS5jb20veGFwLzEuMC8iIHhtcE1NOk9yaWdpbmFsRG9jdW1lbnRJRD0ieG1wLmRpZDpDMkJBNjY5RTU1NEJFMzExOUM4QUM2MDAwNDQzRERBQyIgeG1wTU06RG9jdW1lbnRJRD0ieG1wLmRpZDpCREIzOEIzMzRCN0IxMUUzODhEQjgwOTYzMTgyNTE0QiIgeG1wTU06SW5zdGFuY2VJRD0ieG1wLmlpZDpCREIzOEIzMjRCN0IxMUUzODhEQjgwOTYzMTgyNTE0QiIgeG1wOkNyZWF0b3JUb29sPSJBZG9iZSBQaG90b3Nob3AgQ1M2IChXaW5kb3dzKSI+IDx4bXBNTTpEZXJpdmVkRnJvbSBzdFJlZjppbnN0YW5jZUlEPSJ4bXAuaWlkOkM1QkE2NjlFNTU0QkUzMTE5QzhBQzYwMDA0NDNEREFDIiBzdFJlZjpkb2N1bWVudElEPSJ4bXAuZGlkOkMyQkE2NjlFNTU0QkUzMTE5QzhBQzYwMDA0NDNEREFDIi8+IDwvcmRmOkRlc2NyaXB0aW9uPiA8L3JkZjpSREY+IDwveDp4bXBtZXRhPiA8P3hwYWNrZXQgZW5kPSJyIj8+Af/+/fz7+vn49/b19PPy8fDv7u3s6+rp6Ofm5eTj4uHg397d3Nva2djX1tXU09LR0M/OzczLysnIx8bFxMPCwcC/vr28u7q5uLe2tbSzsrGwr66trKuqqainpqWko6KhoJ+enZybmpmYl5aVlJOSkZCPjo2Mi4qJiIeGhYSDgoGAf359fHt6eXh3dnV0c3JxcG9ubWxramloZ2ZlZGNiYWBfXl1cW1pZWFdWVVRTUlFQT05NTEtKSUhHRkVEQ0JBQD8+PTw7Ojk4NzY1NDMyMTAvLi0sKyopKCcmJSQjIiEgHx4dHBsaGRgXFhUUExIREA8ODQwLCgkIBwYFBAMCAQAAIfkECQcABwAsAAAAACIAFQAAA6J4umv+MDpG6zEj682zsRaWFWRpltoHMuJZCCRseis7xG5eDGp93bqCA7f7TFaYoIFAMMwczB5EkTzJllEUttmIGoG5bfPBjDawD7CsJC67uWcv2CRov929C/q2ZpcBbYBmLGk6W1BRY4MUDnMvJEsBAXdlknk2fCeRk2iJliAijpBlEmigjR0plKSgpKWvEUheF4tUZqZID1RHjEe8PsDBBwkAIfkECQcABwAsAAAAACIAFQAAA6B4umv+MDpG6zEj682zsRaWFWRpltoHMuJZCCRseis7xG5eDGp93TqS40XiKSYgTLBgIBAMqE/zmQSaZEzns+jQ9pC/5dQJ0VIv5KMVWxqb36opxHrNvu9ptPfGbmsBbgSAeRdydCdjXWRPchQPh1hNAQF4TpM9NnwukpRyi5chGjqJEoSOIh0plaYsZBKvsCuNjY5ptElgDyFIuj6+vwcJACH5BAkHAAcALAAAAAAiABUAAAOfeLrc/vCZSaudUY7Nu99GxhhcYZ7oyYXiQQ5pIZgzCrYuLMd8MbAiUu802flYGIhwaCAQDKpQ86nUoWqF6dP00wIby572SXE6vyMrlmhuu9GKifWaddvNQAtszXYCxgR/Zy5jYTFeXmSDiIZGdQEBd06QSBQ5e4cEkE9nnZQaG2J4F4MSLx8rkqUSZBeurhlTUqsLsi60DpZxSWBJugcJACH5BAkHAAcALAAAAAAiABUAAAOgeLrc/vCZSaudUY7Nu99GxhhcYZ7oyYXiQQ5pIZgzCrYuLMd8MbAiUu802flYGIhwaCAQDKpQ86nUoWqF6dP00wIby572SXE6vyMrlmhuu9GuifWaddvNwMkZtmY7AWMEgGcKY2ExXl5khFMVc0Z1AQF3TpJShDl8iASST2efloV5JTyJFpgOch8dgW9KZxexshGNLqgLtbW0SXFwvaJfCQAh+QQJBwAHACwAAAAAIgAVAAADoXi63P7wmUmrnVGOzbvfRsYYXGGe6MmF4kEOaSGYMwq2LizHfDGwIlLPNKGZfi6gZmggEAy2iVPZEKZqzakq+1xUFFYe90lxTsHmim6HGpvf3eR7skYJ3PC5tyystc0AboFnVXQ9XFJTZIQOYUYFTQEBeWaSVF4bbCeRk1meBJYSL3WbaReMIxQfHXh6jaYXsbEQni6oaF21ERR7l0ksvA0JACH5BAkHAAcALAAAAAAiABUAAAOeeLrc/vCZSaudUY7Nu99GxhhcYZ7oyYXiQQ5pIZgzCrYuLMfFlA4hTITEMxkIBMOuADwmhzqeM6mashTCXKw2TVKQyKuTRSx2wegnNkyJ1ozpOFiMLqcEU8BZHx6NYW8nVlZefQ1tZgQBAXJIi1eHUTRwi0lhl48QL0sogxaGDhMlUo2gh14fHhcVmnOrrxNqrU9joX21Q0IUElm7DQkAIfkECQcABwAsAAAAACIAFQAAA6J4umv+MDpG6zEj682zsRaWFWRpltoHMuJZCCRseis7xG5eDGp93bqCA7f7TFaYoIFAMMwczB5EkTzJllEUttmIGoG5bfPBjDawD7CsJC67uWcv2CRov929C/q2ZpcBbYBmLGk6W1BRY4MUDnMvJEsBAXdlknk2fCeRk2iJliAijpBlEmigjR0plKSgpKWvEUheF4tUZqZID1RHjEe8PsDBBwkAIfkECQcABwAsAAAAACIAFQAAA6B4umv+MDpG6zEj682zsRaWFWRpltoHMuJZCCRseis7xG5eDGp93TqS40XiKSYgTLBgIBAMqE/zmQSaZEzns+jQ9pC/5dQJ0VIv5KMVWxqb36opxHrNvu9ptPfGbmsBbgSAeRdydCdjXWRPchQPh1hNAQF4TpM9NnwukpRyi5chGjqJEoSOIh0plaYsZBKvsCuNjY5ptElgDyFIuj6+vwcJACH5BAkHAAcALAAAAAAiABUAAAOfeLrc/vCZSaudUY7Nu99GxhhcYZ7oyYXiQQ5pIZgzCrYuLMd8MbAiUu802flYGIhwaCAQDKpQ86nUoWqF6dP00wIby572SXE6vyMrlmhuu9GKifWaddvNQAtszXYCxgR/Zy5jYTFeXmSDiIZGdQEBd06QSBQ5e4cEkE9nnZQaG2J4F4MSLx8rkqUSZBeurhlTUqsLsi60DpZxSWBJugcJACH5BAkHAAcALAAAAAAiABUAAAOgeLrc/vCZSaudUY7Nu99GxhhcYZ7oyYXiQQ5pIZgzCrYuLMd8MbAiUu802flYGIhwaCAQDKpQ86nUoWqF6dP00wIby572SXE6vyMrlmhuu9GuifWaddvNwMkZtmY7AWMEgGcKY2ExXl5khFMVc0Z1AQF3TpJShDl8iASST2efloV5JTyJFpgOch8dgW9KZxexshGNLqgLtbW0SXFwvaJfCQAh+QQJBwAHACwAAAAAIgAVAAADoXi63P7wmUmrnVGOzbvfRsYYXGGe6MmF4kEOaSGYMwq2LizHfDGwIlLPNKGZfi6gZmggEAy2iVPZEKZqzakq+1xUFFYe90lxTsHmim6HGpvf3eR7skYJ3PC5tyystc0AboFnVXQ9XFJTZIQOYUYFTQEBeWaSVF4bbCeRk1meBJYSL3WbaReMIxQfHXh6jaYXsbEQni6oaF21ERR7l0ksvA0JACH5BAkHAAcALAAAAAAiABUAAAOeeLrc/vCZSaudUY7Nu99GxhhcYZ7oyYXiQQ5pIZgzCrYuLMfFlA4hTITEMxkIBMOuADwmhzqeM6mashTCXKw2TVKQyKuTRSx2wegnNkyJ1ozpOFiMLqcEU8BZHx6NYW8nVlZefQ1tZgQBAXJIi1eHUTRwi0lhl48QL0sogxaGDhMlUo2gh14fHhcVmnOrrxNqrU9joX21Q0IUElm7DQkAOw==) ; width: 36px !important; height: 21px !important; border: none !important; float: right; margin-top: -7px; margin-right: -10px; transform: scale(2,1); } .fullimg { height: 550px !important; } .reveal section img { background: white; } </style>  # RabbitMQ @Kais(VagrantPi) ###### tags: `slide`, `簡報`, `RabbitMQ` ---  ## Agenda - Hello World! - Work queues - Publish/Subscribe - Routing - Topics - RPC - sample code link ---  ### Hello World! ![](https://www.rabbitmq.com/img/tutorials/python-one.png) ----  send.js(producer) ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var queue = 'hello'; var msg = 'Hello World!'; // 先戳一下 queue，如果不存在就建立 // 也可以在建立 queue 時帶入參數 // durable: if true, the queue will survive broker restarts channel.assertQueue(queue, { durable: false }); channel.sendToQueue(queue, Buffer.from(msg)); console.log(" [x] Sent %s", msg); }); setTimeout(function() { connection.close(); process.exit(0); }, 500); }); ``` ----  receive.js(consumers) ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var queue = 'hello'; // 需要注意的是，所有 assertQueue 'hello' queue 帶的參數要一樣，不然會噴 error channel.assertQueue(queue, { durable: false }); console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue); channel.consume(queue, function(msg) { console.log(" [x] Received %s", msg.content.toString()); }, { noAck: true }); }); }); ``` ---  ### Work queues ![](https://www.rabbitmq.com/img/tutorials/python-two.png) ``` 預設值情況下，RabbitMQ 會輪著分配，所以在偶數 tack 情況下，每個 worker 分配到的工作是一樣的，這稱為：round-robin ``` ----  #### acknowledgment ```javascript noAck: true channel.consume(queue, function(msg) { .... }, {noAck: false});// 開啟 acknowledgment ``` ``` 當 worker 拿到 task 掛掉後，RabbitMQ 可以透過 acknowledgment 來發現，並將 task 在分派下去，原理及 worker 接收到後向 RabbitMQ 發送訊息所已經取得了 `channel.ack(msg);` ``` ----  #### Message durability ``` 上面的 acknowledgment 使 consumer 掛掉後，任務不會遺失，但假設 RabbitMQ 主機掛掉，task 還是會遺失，因此需要設定 queue 為 durable ，然後傳送的 message 必須設定 persistent ``` ```javascript channel.assertQueue（'hello'，{ durable：true }）; ``` ``` 但是，assertQueue 原本以建立的 Queue 餵的參數不能跟當初建立時的不同 ``` ```javascript channel.sendToQueue(q, Buffer.from(msg), {persistent: true}); ``` ----  #### Fair dispatch ![](https://www.rabbitmq.com/img/tutorials/prefetch-count.png) ```javascript channel.prefetch(1); ``` ``` 前面有提到預設採用 round-robin 的分配 task 方式，所以不管 worker 是否還在執行 task，他還是會往他那邊塞，prefetch=1 的意思為 worker 同一時間只會接收一個 task ``` ----  new_task.js(producer) ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var queue = 'task_queue'; var msg = process.argv.slice(2).join(' ') || "Hello World!"; channel.assertQueue(queue, { durable: true }); channel.sendToQueue(queue, Buffer.from(msg), { persistent: true }); console.log(" [x] Sent '%s'", msg); }); setTimeout(function() { connection.close(); process.exit(0); }, 500); }); ``` ----  worker.js(consumers) ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error, connection) { connection.createChannel(function(error, channel) { var queue = 'task_queue'; channel.assertQueue(queue, { durable: true }); channel.prefetch(1); console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue); channel.consume(queue, function(msg) { var secs = msg.content.toString().split('.').length - 1; console.log(" [x] Received %s", msg.content.toString()); setTimeout(function() { console.log(" [x] Done"); channel.ack(msg); }, secs * 1000); }, { noAck: false }); }); }); ``` ---  ### Publish/Subscribe ![](https://www.rabbitmq.com/img/tutorials/exchanges.png) ``` 要實作一個 log 記錄器，在寫入硬碟的同時(worker1)，並輸出至螢幕(worker2) ``` ----  #### Exchanges ``` 這邊在架構上引入一個新角色 - Exchanges，producer 透過 Exchanges 分排到你預期的 queue 中，可以使其更加彈性，缺點則是當單一 Exchanges 故障，正個 service 就會停擺，Exchanges 提供 direct, topic, headers, fanout 不同模式，該範例以 fanout 為主 ``` ----  #### Temporary queues 對於一些需求上，可能不需要明確的建立 queue（只需要暫存的） ```javascript channel.assertQueue（''，{ // exclusive 為 true 時，只會讓一個 connection 連接，並在結束時會刪除該 queue exclusive：true }）; ``` ----  #### Bindings ![](https://www.rabbitmq.com/img/tutorials/bindings.png) ``` 建立完 Exchange 和 Queue 後，需要綁定兩者之間的關係 ``` ----  #### Putting it all together ![](https://www.rabbitmq.com/img/tutorials/python-three-overall.png) ----  emit_log.js ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var exchange = 'logs'; var msg = process.argv.slice(2).join(' ') || 'Hello World!'; channel.assertExchange(exchange, 'fanout', { durable: false }); channel.publish(exchange, '', Buffer.from(msg)); console.log(" [x] Sent %s", msg); }); setTimeout(function() { connection.close(); process.exit(0); }, 500); }); ``` ----  receive_logs.js ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var exchange = 'logs'; channel.assertExchange(exchange, 'fanout', { durable: false }); channel.assertQueue('', { exclusive: true }, function(error2, q) { if (error2) { throw error2; } console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q.queue); // channel.bindQueue(queue, source, pattern, [args, [function(err, ok) {...}]]) channel.bindQueue(q.queue, exchange, ''); channel.consume(q.queue, function(msg) { if (msg.content) { console.log(" [x] %s", msg.content.toString()); } }, { noAck: true }); }); }); }); ``` ----  ``` ./receive_logs.js > logs_from_rabbit.log ``` ``` ./receive_logs.js ``` ``` ./emit_log.js ``` ---  ### Routing ![](https://www.rabbitmq.com/img/tutorials/python-four.png) ----  #### Bindings ```javascript channel.bindQueue(q.queue, exchange, ''); ``` bindQueue 可以簡單理解成該 queue 只對什麼 exchanges 訊息有興趣 ----  #### Direct exchange ![](https://www.rabbitmq.com/img/tutorials/direct-exchange.png) message(routing key) -> direct + binding key ----  #### Multiple bindings ![](https://www.rabbitmq.com/img/tutorials/direct-exchange-multiple.png) ----  emit_log_direct.js ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var exchange = 'direct_logs'; var args = process.argv.slice(2); var msg = args.slice(1).join(' ') || 'Hello World!'; var severity = (args.length > 0) ? args[0] : 'info'; channel.assertExchange(exchange, 'direct', { durable: false }); channel.publish(exchange, severity, Buffer.from(msg)); console.log(" [x] Sent %s: '%s'", severity, msg); }); setTimeout(function() { connection.close(); process.exit(0) }, 500); }); ``` ----  receive_logs_direct.js ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); var args = process.argv.slice(2); if (args.length == 0) { console.log("Usage: receive_logs_direct.js [info] [warning] [error]"); process.exit(1); } amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var exchange = 'direct_logs'; channel.assertExchange(exchange, 'direct', { durable: false }); channel.assertQueue('', { exclusive: true }, function(error2, q) { if (error2) { throw error2; } console.log(' [*] Waiting for logs. To exit press CTRL+C'); args.forEach(function(severity) { channel.bindQueue(q.queue, exchange, severity); }); channel.consume(q.queue, function(msg) { console.log(" [x] %s: '%s'", msg.fields.routingKey, msg.content.toString()); }, { noAck: true }); }); }); }); ``` ----  ![](https://www.rabbitmq.com/img/tutorials/python-four.png) ``` ./receive_logs_direct.js warning error > logs_from_rabbit.log ``` ``` ./receive_logs_direct.js info warning error ``` ``` ./emit_log_direct.js error "Run. Run. Or it will explode." ``` ---  ### Topics ![](https://www.rabbitmq.com/img/tutorials/python-five.png) ----  又回到我們的 log 服務上，log 不只有各種 severity level 的差別 (info/warn/crit...) 也會有其他像 facility (auth/cron/kern...). ----  #### Topic exchange ``` 設定廣播的 topic 都用 `.` 隔開，ex: "stock.usd.nyse", "nyse.vmw" , "quick.orange.rabbit(最多可到 255 bytes) binding 的方式也跟 direct 很像，選定要訂閱的 topic 就好 - `*` 可代替一個單字 - `#` 可代替 0 或多個單字 ex: *.orange.*", "*.*.rabbit", "lazy.#" ``` ----  範例 ``` Queue1: "*.orange.*" Queue2: "*.*.rabbit", "lazy.#" routing key: "quick.orange.rabbit" ※ 如果送出的 lazy.orange.male.rabbit 這種長度跟 "*.orange.*", "*.*.rabbit" 的合，不過跟 "lazy.#" 有 match，所以只會傳到 Queue2 ``` ----  emit_log_topic.js ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var exchange = 'topic_logs'; var args = process.argv.slice(2); var key = (args.length > 0) ? args[0] : 'anonymous.info'; var msg = args.slice(1).join(' ') || 'Hello World!'; channel.assertExchange(exchange, 'topic', { durable: false }); channel.publish(exchange, key, Buffer.from(msg)); console.log(" [x] Sent %s:'%s'", key, msg); }); setTimeout(function() { connection.close(); process.exit(0) }, 500); }); ``` ----  receive_logs_topic.js: ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); var args = process.argv.slice(2); if (args.length == 0) { console.log("Usage: receive_logs_topic.js <facility>.<severity>"); process.exit(1); } amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var exchange = 'topic_logs'; channel.assertExchange(exchange, 'topic', { durable: false }); channel.assertQueue('', { exclusive: true }, function(error2, q) { if (error2) { throw error2; } console.log(' [*] Waiting for logs. To exit press CTRL+C'); args.forEach(function(key) { channel.bindQueue(q.queue, exchange, key); }); channel.consume(q.queue, function(msg) { console.log(" [x] %s:'%s'", msg.fields.routingKey, msg.content.toString()); }, { noAck: true }); }); }); }); ``` ---  ### Remote procedure call (RPC) ![](https://www.rabbitmq.com/img/tutorials/python-six.png) ----  #### Callback queue ```javascript channel.assertQueue('', { exclusive: true }); channel.sendToQueue('rpc_queue', Buffer.from('10'), { replyTo: queue_name }); # ... then code to read a response message from the callback queue ... ``` ----  #### Message properties ``` AMQP 0-9-1 protocol predefines a set of 14 properties * persistent: tutorial 2 有提到，可以確保 restart 後資料還在 * content_type: mime-type，ex: `application/json`. * reply_to: callback queue 的名字 * correlation_id: 識別 Useful to correlate RPC responses with requests. ``` ----  #### Correlation Id 如果每次的 RPC callback 都建立一個 callback queue 這樣沒什麼效率，所以好的方法是只建立一個 callback 的 queue，不過為了分辨這次 callback 是哪一次哪一個 producer 傳的，所以需要一個 Correlation Id 來做識別 ----  rpc_client.js ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); var args = process.argv.slice(2); if (args.length === 0) { console.log("Usage: rpc_client.js num"); process.exit(1); } amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } channel.assertQueue('', { exclusive: true }, function(error2, q) { if (error2) { throw error2; } var correlationId = generateUuid(); var num = parseInt(args[0]); console.log(' [x] Requesting fib(%d)', num); channel.consume(q.queue, function(msg) { if (msg.properties.correlationId === correlationId) { console.log(' [.] Got %s', msg.content.toString()); setTimeout(function() { connection.close(); process.exit(0); }, 500); } }, { noAck: true }); channel.sendToQueue('rpc_queue', Buffer.from(num.toString()), { correlationId: correlationId, replyTo: q.queue }); }); }); }); function generateUuid() { return Math.random().toString() + Math.random().toString() + Math.random().toString(); } ``` ----  ```javascript #!/usr/bin/env node var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var queue = 'rpc_queue'; channel.assertQueue(queue, { durable: false }); channel.prefetch(1); console.log(' [x] Awaiting RPC requests'); channel.consume(queue, function reply(msg) { var n = parseInt(msg.content.toString()); console.log(" [.] fib(%d)", n); var r = fibonacci(n); channel.sendToQueue(msg.properties.replyTo, Buffer.from(r.toString()), { correlationId: msg.properties.correlationId }); channel.ack(msg); }); }); }); function fibonacci(n) { if (n === 0 || n === 1) return n; else return fibonacci(n - 1) + fibonacci(n - 2); } ``` ---  ## sample code - [sample code](https://bitbucket.org/kaislin/bolt-queue/src/master/) - [官方 tutorials code](https://github.com/rabbitmq/rabbitmq-tutorials/tree/master/javascript-nodejs) - [node.js amqplib sample code](https://github.com/squaremo/amqp.node/tree/master/examples/tutorials)

Cheatsheet

Syntax	Example	Reference
# Header	Header	基本排版
- Unordered List	Unordered List
1. Ordered List	Ordered List
- [ ] Todo List	Todo List
> Blockquote	Blockquote
Bold font	Bold font
Italics font	Italics font
~~Strikethrough~~	~~Strikethrough~~
19^th^	19^th
H~2~O	H₂O
++Inserted text++	Inserted text
==Marked text==	Marked text
[link text](https:// "title")	Link
![image alt](https:// "title")	Image
`Code`	`Code`	在筆記中貼入程式碼
```javascript var i = 0; ```	`var i = 0;`	在筆記中貼入程式碼
:smile:		Emoji list
{%youtube youtube_id %}	Externals
$L^aT_eX$	L^aT_eX
:::info This is a alert area. :::	This is a alert area.