摘要:
使用RabbitMQ的消息队列,可以有效提高系统的峰值处理能力。
RabbitMQ简介
RabbitMQ是
消息代理(Message Broker)
,它支持多种异步消息处理方式,最常见的有:
RabbitMQ还支持Routing、Topics、以及Remote procedure calls (RPC)等方式。
对于不同的消息处理方式,有一点是相同的,RabbitMQ是介于消息的生产者和消费者的中间节点,负责缓存和分发消息。RabbitMQ接收来自生产者的消息,缓存到内存中,按照不同的方式分发给消费者。RabbitMQ还可以将消息写入磁盘,保证持久化,这样即使RabbitMQ意外崩溃了,消息数据不至于完全丢失。
为什么使用RabbitMQ?
最简单的一点在于,它支持Work Queue等不同的消息处理方式,可以用于不同的业务场景。对于我们Fundebug来说,目前只用过RabbitMQ的Work Queue,即消息队列。
使用消息队列,可以将不算紧急、但是非常消耗资源的计算任务,以消息的方式插入到RabbitMQ的队列中,然后使用多个处理模块处理这些消息。
这样做最大的好处在于:
提高了系统峰值处理能力
。因为,来不及处理的消息缓存在RabbitMQ中,避免了同时进行大量计算导致系统因超负荷运行而崩溃。而那些来不及处理的消息,会在峰值过去之后慢慢处理掉。
另一个好处在于
解耦
。消息的生产者只需要将消息发送给RabbitMQ,这些消息什么时候处理完,不会影响生产者的响应性能。
广告:欢迎免费试用Fundebug,为您监控线上代码的BUG,提高用户体验~
安装并运行RabbitMQ
使用Docker运行RabbitMQ非常简单,只需要执行一条简单的命令:
sudo docker run -d --name rabbitmq -h rabbitmq -p 5672:5672 -v /var/lib/rabbitmq:/var/lib/rabbitmq registry.docker-cn.com/library/rabbitmq:3.7
|
对于不熟悉Docker的朋友,我解释一下docker的命令选项:
-
-d
: 后台运行容器
-
–name rabbitmq
: 将容器的名字设为rabbitmq
-
-h rabbitmq
: 将容器的主机名设为rabbitmq,希望RabbitMQ消息数据持久化保存到本地磁盘是需要设置主机名,因为RabbitMQ保存数据的目录为主机名
-
-p 5672:5672
: 将容器的5672端口映射为本地主机的5672端口,这样可以通过本地的5672端口访问rabbitmq
-
-v /var/lib/rabbitmq:/var/lib/rabbitmq
:将容器的/var/lib/rabbitmq目录映射为本地主机的/var/lib/rabbitmq目录,这样可以将RabbitMQ消息数据持久化保存到本地磁盘,即使RabbitMQ容器被删除,数据依然还在。
Docker为官方镜像提供了加速服务,因此命令中Rabbit的Docker镜像名为
registry.docker-cn.com/library/rabbitmq:3.7
。
如果你不会Docker,建议你学习一下。如果你不想学,Ubuntu 14.04下安装RabbitMQ的命令是这样的:
sudo echo "deb http://www.rabbitmq.com/debian testing main" | sudo tee -a /etc/apt/sources.list wget -O- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add - sudo apt-get update sudo apt-get install rabbitmq-server
|
启动RabbitMQ:
sudo service rabbitmq-server start
|
消息队列代码示例
下面,我们使用Node.js实现一个简单消息队列。
消息的生产者:sender.js
const amqp = require("amqplib");
const queue = "demo";
async function sendMessage(message)
{ const connection = await amqp.connect("amqp://localhost"); const channel = await connection.createChannel(); await channel.assertQueue(queue); await channel.sendToQueue(queue, new Buffer(message), { persistent: true }); }
setInterval(function() { sendMessage("Hello, Fundebug!"); }, 1000)
|
消息的消费者:receiver.js
const amqp = require("amqplib");
const queue = "demo";
async function receiveMessage() { const connection = await amqp.connect("amqp://localhost"); const channel = await connection.createChannel(); await channel.assertQueue(queue); await channel.consume(queue, function(message) { console.log(message.content.toString()); channel.ack(message); }); }
receiveMessage();
|
我们用到了amqplib模块,用于与RabbitMQ进行通信,对于具体接口的细节,可以查看文档。
在调用sendToQueue时,将persistent属性设为true,这样RabbitMQ关闭时,消息会被保存到磁盘。测试这一点很简单:
由于RabbitMQ容器将保存数据的目录(/var/lib/rabbitmq)以数据卷的形式保存在本地主机,因此即使将RabbitMQ容器删除(sudo docker rm -f rabbitmq)后重新运行,效果也是一样的。
另外,这段代码采用了Node.js最新的异步代码编写方式:Async/Await,因此非常简洁,感兴趣的同学可以了解一下。
这个Demo的运行方式非常简单:
在receiver端,可以看到不停地打印”Hello, Fundebug!”。
代码仓库地址为:Fundebug/rabbitmq-demo
自动重连代码示例
在生产环境中,RabbitMQ难免会出现重启的情况,比如更换磁盘或者服务器、负载过高导致崩溃。因为RabbitMQ可以将消息写入磁盘,所以数据是”安全”的。但是,代码中必须实现自动重连机制,否则RabbitMQ停止时会导致Node.js应用崩溃。这里提供一个自动重连的代码示例,给大家参考:
消息生产者:sender_reconnect.js
const amqp = require("amqplib");
const queue = "demo";
var connection;
async function connectRabbitMQ() { try { connection = await amqp.connect("amqp://localhost"); console.info("connect to RabbitMQ success");
const channel = await connection.createChannel(); await channel.assertQueue(queue); await channel.sendToQueue(queue, new Buffer("Hello, Fundebug!"), { persistent: true });
connection.on("error", function(err) { console.log(err); setTimeout(connectRabbitMQ, 10000); });
connection.on("close", function() { console.error("connection to RabbitQM closed!"); setTimeout(connectRabbitMQ, 10000); });
} catch (err) { console.error(err); setTimeout(connectRabbitMQ, 10000); } }
connectRabbitMQ();
|
消息消费者:receiver_reconnect.js
const amqp = require("amqplib");
const
|