RabbitMq简介

RabbitMq 是一个消息代理，允许在不同服务之间发送和接收消息。它是一个实现高级消息队列协议（AMQP）的消息代理。用 Erlang 编程语言编写。

安装 RabbitMq

RabbitMq 可以使用各自的包管理器安装在不同的操作系统上。 RabbitMQ需要预先安装Erlang/OTP，更多信息请参考官方文档。

• Linux

sudo apt-get install rabbitmq-server

• 苹果

brew install rabbitmq

安装完成后，我们可以使用以下命令启动RabbitMq服务器：

sudo rabbitmq-server

服务器启动后，我们可以使用默认端口 5672 从我们的应用程序访问rabbitmq。

要启用RabbitMq管理控制台，我们可以运行以下命令：

sudo rabbitmq-plugins enable rabbitmq_management

我们可以通过浏览器访问http://localhost:15672/来访问RabbitMq管理控制台。默认用户名和密码为 guest。

使用 NodeJs 设置 RabbitMq

要从 NodeJs 与 RabbitMq 交互，我们可以使用 amqplib 库。我们可以使用以下命令安装该库：

npm install amqplib

要从NodeJs连接到RabbitMq并创建与RabbitMq交互的通道，我们可以使用以下代码：

const amqp = require("amqplib");

const connect = async (url) => {
  const connection = await amqp.connect(url);
  const channel = await connection.createChannel();

  return channel;
};

connect("amqp://localhost:5672").then((channel) => {
  console.log("Connected to RabbitMQ");
  channel.close();
  process.exit(0);
});

现在，我们有了一个与 RabbitMq 交互的通道。我们可以使用此通道从 RabbitMq 发送和接收消息。让我们将队列命名为 mq-test-queue 并向队列发送一条消息。

connect("amqp://localhost:5672").then((channel) => {
  console.log("Connected to RabbitMQ");
  channel.assertQueue("mq-test-queue");
  console.log("Queue created");
  channel.sendToQueue("mq-test-queue", Buffer.from("Hello World!"));
  console.log("Message sent");
});

我们创建了一个队列 mq-test-queue 并发送了一条消息 Hello World！到队列中。为了创建队列，我们​​使用了assertQueue方法，为了向队列发送消息，我们使用了sendToQueue方法。

要从队列接收消息，我们可以使用consume方法。

connect("amqp://localhost:5672").then((channel) => {
  console.log("Connected to RabbitMQ");
  channel.assertQueue("mq-test-queue");
  console.log("Queue created");
  channel.sendToQueue("mq-test-queue", Buffer.from("Hello World!"));
  console.log("Message sent");
  channel.consume("mq-test-queue", (msg) => {
    console.log("Message received", msg.content.toString());
  });
});

consum 方法接受一个回调函数，并将消息对象传递给回调函数，以使用 msg.content.toString().

消费消息后，我们需要ack或nack消息，以将消息从队列中移除。我们可以使用 ack 方法来确认消息，使用 nack 方法来拒绝消息。

对于这个例子，我们将在消费消息后确认消息。

const amqp = require("amqplib");

const connect = async (url) => {
  const connection = await amqp.connect(url);
  const channel = await connection.createChannel();

  return channel;
};

connect("amqp://localhost:5672").then((channel) => {
  console.log("Connected to RabbitMQ");
  channel.assertQueue("mq-test-queue");
  console.log("Queue created");
  channel.sendToQueue("mq-test-queue", Buffer.from("Hello World!"));
  console.log("Message sent");
  channel.consume("mq-test-queue", (msg) => {
    console.log("Message received", msg.content.toString());
    channel.ack(msg);
  });
});

我们可以看到队列中的消息发送和接收成功。使用rabbitmq管理控制台。

结论

在本文中，我们学习了如何使用 amqplib 库设置 RabbitMq 并从 NodeJs 与 RabbitMq 交互。我们创建了一个队列，向队列发送了一条消息，并从队列中接收了一条消息。