cpselvis

落花无言,人淡如菊,心素如简

嗨,我是程柳锋 (@cpselvis),一名来自中国的 web 开发者。现居深圳,就职于 腾讯。目前负责研发规范和工程化建设,曾在OSC源创会上做过分享。


在Node.js中使用RabbitMQ系列一 Hello world

本篇文章以Node.js为例子,来实际操作如何和RabbitMQ进行交互。

介绍

RabbitMQ是一个消息代理。它最初的思想特别简单:接受并且转发消息。你可以将它想象为邮局:当你将邮件放到信箱中,你可以非常肯定快件员最终会将邮件交到接受人手中。你可以把RabbitMQ比喻为信箱、邮局和快递员。RabbitMQ和邮局之间主要的区别是它不处理纸张,而是接受、存储和转发二进制数据‒消息。

在RabbitMQ中,有一些基本术语:

  • 生产者:就是发送信息这方。
  • 任务队列:虽然信息流在RabbitMQ和你的应用之间流动,它可以存储在一个队列中。队列不受任何限制的约束,它可以存储任意多的消息,它本质上是一个无限的缓冲区。许多生产者可以将消息发往到这个队列中,许多消费者可以尝试从这个队列中接受数据。
  • 消费者:消费者和信息接受者有相近的含义,一个消费者就是一个等待去接受信息的程序。

安装RabbitMQ server

打开RabbitMQ的下载页面(https://www.rabbitmq.com/download.html),下载安装,这里以Mac OSX平台安装为例: RabbitMQ依赖Erlang,由于Mac OSX本身已经安装了Erlang,所以可以直接通过Homebrew来进行安装。

$ brew update
$ brew install rabbitmq

安装完后,需要将/usr/local/sbin添加到$PATH,添加到./.bash_profile文件,然后

$ source ./.bash_profile
$ echo $PATH      // 检查环境变量是否已经成功加入

安装完成后就可以启动RabbitMQ server了。

至此,安装就完成了。运行rabbitmq-server命令时可能会报错误:ERROR: epmd error for host localhost: timeout (timed out),如果遇到这种情况,可以打开/etc/hosts文件,在文件末尾加上 127.0.0.1 localhost即可解决问题。

Hello world

在这个部分,我会使用Javascript编写两个小程序。一个发送单个消息的生产者和接收消息并将其打印出来的消费者。我们将跳过在amqp.node API的一些细节,集中在这个非常简单的事情。

在下图中,P代表生产者,C代表消费者,中间红色代表的是任务队列-消息缓冲区

首先,使用npm安装amqp.node

$ npm install amqplib

发送消息

这里我将消息的发送者称作send.js,消息接受者称作receive.js,消息发送者会连接到RabbitMQ,发送一个消息,最后退出。 首先引入amqplib这个模块:

var amqp = require('amqplib/callback_api');

连接到 RabbitMQ server

amqp.connect('amqp://localhost', function(err, conn) {});

接下来创建一个通道,

amqp.connect('amqp://localhost', function(err, conn) {
  conn.createChannel(function(err, ch) {});
});

为了发送消息,我们需要定义一个队列,我们可以将消息发送到这个队列中:

amqp.connect('amqp://localhost', function(err, conn) {
  conn.createChannel(function(err, ch) {
    var q = 'hello';

    ch.assertQueue(q, {durable: false});
    // Note: on Node 6 Buffer.from(msg) should be used
    ch.sendToQueue(q, new Buffer('Hello World!'));
    console.log(" [x] Sent 'Hello World!'");
  });
});

最后,我们关闭连接,并且退出:

setTimeout(function() { conn.close(); process.exit(0) }, 500);

最终代码参考:send.js

接受消息

建立一个接受者的方式和发送者是相同的。打开一个连接和通道,并且申明一个需要处理的队列。注意:这里的队列和发送者里面定义的队列需要匹配。

amqp.connect('amqp://localhost', function(err, conn) {
  conn.createChannel(function(err, ch) {
    var q = 'hello';

    ch.assertQueue(q, {durable: false});
  });
});

这里也定义队列的原因:接受者可能比发送者先开始执行。我们需要确保当接受者处理queue的时候,queue是存在的。 由于消息的发送是异步的,我们需要提供一个回调,这样,当RabbitMQ发送消息给我们的消费者时,回调会执行。这个也是Channel.consume做的事情。

console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q);
ch.consume(q, function(msg) {
  console.log(" [x] Received %s", msg.content.toString());
}, {noAck: true});

最终代码参考:receive.js

运行代码

// 先执行send.js
$ ./send.js
// 后执行receive.js
$ ./receive.js

最终结果如图:

开源信息

最近的文章

在Node.js中使用RabbitMQ系列二 任务队列

在上一篇文章在Node.js中使用RabbitMQ系列一 Hello world我有使用一个任务队列,不过当时的场景是将消息发送给一个消费者,本篇文章我将讨论有多个消费者的场景。其实,任务队列最核心解决的问题是避免立即处理那些耗时的任务,也就是避免请求-响应的这种同步模式。取而代之的是我们通过调度算法,让这些耗时的任务之后再执行,也就是采用异步的模式。我们需要将一条消息封装成一个任务,并且将它添加到任务队列里面。后台会运行多个工作进程(worker process),通过调度算法,将队列里...…

继续阅读
更早的文章

理解Javascript的状态容器Redux

Redux要解决什么问题?随着 JavaScript 单页应用开发日趋复杂,JavaScript 需要管理比任何时候都要多的 state (状态)。 这些 state 可能包括服务器响应、缓存数据、本地生成尚未持久化到服务器的数据,也包括 UI 状态,如激活的路由,被选中的标签,是否显示加载动效或者分页器等等。管理不断变化的 state 非常困难。如果一个 model 的变化会引起另一个 model 变化,那么当 view 变化时,就可能引起对应 model 以及另一个 model 的变化...…

继续阅读