Heroku上的Node JS消息队列

Question

我需要将在Heroku上运行的Node JS服务器转移到消息队列架构。 目前，服务器接收HTTP请求，进行一些处理并响应。 问题在于处理需要一些时间，特别是当有很多请求时。 这种漫长的处理时间会导致服务器超时、过载和崩溃！ 我的阅读告诉我需要一个后台工作者来处理这些。

我对消息队列和后台工作者毫无经验，我正在寻找一个非常简单的例子来入门。 有人能建议一个简单且易懂的模块或示例吗？

我找到了一些示例，但它们很复杂，我迷失了！ 我想要一个最基本的例子，以便我可以从中构建。

- user3320795

---

酷。你有Screencast的任何预览/样本可以看吗？我想看看这是不是正确的水平。你所教授的内容是否适用于Heroku？ - user3320795

其中一个采访是免费的，链接在这里：https://sub.watchmecode.net/episode/rmq-interviews-udi-dahan/ 但我没有太多的预览内容，否则...我应该制作一个预览视频。 - Derick Bailey

Answer 1

让我们看看如何使用RabbitMQ完成这个任务。 首先，在您的开发环境中需要一个RabbitMQ服务器来使用。 如果您还没有它（检查“sudo service rabbitmq-server status”），您可以按照以下方式安装（在ubuntu或类似系统中）：

sudo su -c "echo 'deb http://www.rabbitmq.com/debian/ testing main' >> /etc/apt/sources.list"
wget http://www.rabbitmq.com/rabbitmq-signing-key-public.asc
sudo apt-key add rabbitmq-signing-key-public.asc
sudo apt-get update
sudo apt-get install rabbitmq-server
rm  rabbitmq-signing-key-public.asc

然后，使用以下命令启动服务器：

sudo service rabbitmq-server start

你还需要为Heroku部署配置一个RabbitMQ服务。我们以CloudAMPQ为例。您可以使用以下内容将其免费计划添加到Heroku应用程序中：

您还需要为Heroku部署配置一个RabbitMQ服务。让我们以CloudAMPQ为例。您可以通过以下步骤将其免费计划添加到Heroku应用中：

heroku addons:create cloudamqp:lemur 

这将在您的Heroku应用程序中创建一个新的CLOUDAMQP_URL环境变量。

接下来，您需要为node.js应用程序选择一个合适的RabbitMQ客户端。 有几个可以选择，但是在此示例中，让我们使用ampqlib：

npm install ampqlib --save

这应该在您的 package.json 的 dependencies 中添加类似以下行：

"amqplib": "^0.4.1",

下一步是在您的Heroku应用程序中添加一个后台“工作者”dyno。 我假设您当前的Procfile中只有一个Web dyno。 因此，您需要添加另一行来实例化一个工作进程，例如：

worker: node myworker.js

最后，您需要编写代码，使您的Web dyno能够通过RabbitMQ与您的worker dyno进行交互。

为了举例说明，我假设您的Web dyno将向RabbitMQ消息队列“发布”消息，而您的worker dyno将“消费”这些消息。

因此，让我们从编写发布消息到消息队列的代码开始。此代码需要在您的Web dyno的某个位置运行：

// Define ampq_url to point to CLOUDAMPQ_URL on Heroku, or local RabbitMQ server in dev environment
var ampq_url = process.env.CLOUDAMQP_URL || "amqp://localhost";
var ampq_open = require('amqplib');
var publisherChnl;

function createPublisherChannel() {

    // Create an AMPQ "connection"
    ampq_open.connect(ampq_url)
        .then(function(conn) {
            // You need to create at least one AMPQ "channel" on your connection   
            var ok = conn.createChannel();
            ok = ok.then(function(ch){
                publisherChnl = ch;
                // Now create a queue for the actual messages to be sent to the worker dyno 
                publisherChnl.assertQueue('my-worker-q');
            })
        })
    }

function publishMsg() {
     // Send the worker a message
     publisherChnl.sendToQueue('my-worker-q', new Buffer('Hello world from Web dyno'));
}

您需要在Web dyno的初始化过程中调用createPublisherChannel()。每当您想向队列发送消息时，请调用publishMsg()。

最后，让我们编写代码以在worker dyno中消耗上述消息。例如，在myworker.js中添加类似以下内容的内容：

// Just like in Web dyno...
var amqp_url = process.env.CLOUDAMQP_URL || "amqp://localhost";
var open_ampq = require('amqplib').connect(amqp_url);
var consumerChnl;    

// Creates an AMPQ channel for consuming messages on 'my-worker-q'
function createConsumerChannel() {     
    open_ampq
        .then(function(conn) {
            conn.createChannel()
                .then(function(ch) {
                    ch.assertQueue('my-worker-q');
                    consumerChnl = ch;
            });
        });
}  

function startConsuming() {
    consumerChnl.consume('my-worker-q', function(msg){
        if (msg !== null) {
            console.log(msg.content.toString());
            // Tell RabbitMQ server we have consumed the message
            consumerChnl.ack(msg);
        }
    })
} 

createConsumerChnl().then(startConsuming); 

最后，使用 "heroku local" 进行测试。您应该可以看到您的应用现在有两个进程正在运行，“Web”和“worker”。每当您在 Web dyno 中调用 publishMsg() 时，您希望可以在 wroker dyno 中看到消息内容被输出到控制台。要查看 RabbitMQ 队列中发生的情况，可以使用以下命令：

sudo rabbitmqctl list_queues

- Yoni Rabinovitch

---

谢谢您的解释，非常有帮助。不过我有一个问题.. 我该如何在不同的模块之间共享这个 publisherChnl？ - Kamila

我猜你的意思是想要在Web应用程序中的多个Node.js模块中发布消息到队列上？ 如果是这样，只需导出函数publishMsg并使用“exports.publishMsg = publishMsg”，这样你就可以从任何地方调用它。 如果不是这个意思，请澄清一下。 - Yoni Rabinovitch

是的，那是我的问题。感谢您的澄清。我还有一个疑问..我尝试了这段代码，看起来startConsuming()在createConsumerChannel()完成之前就已经运行了。最好的链接方式是什么？我可以通过在consumerChnl = ch;后立即调用startConsuming()来使其工作，但这是正确的方法吗？ - Kamila

Answer 2

我在这里找到了一个非常简单的例子（后面还有更深入的例子）：https://www.rabbitmq.com/tutorials/tutorial-one-javascript.html

- user3320795