笔记大全

浅析node中怎么使用消息队列

更新时间：2023-07-04

什么是消息队列

消息队列是一种在分布式系统中用于应用程序之间进行异步通信的技术。它将消息发送到一个或多个队列中，并将消息存储在队列中，直到接收方准备好处理它们。这种方式允许应用程序之间解耦，提高了系统的可伸缩性和可靠性。在Node中，我们可以使用Redis，RabbitMQ和Kafka等消息队列中间件来实现消息队列的功能。

使用Redis作为消息队列

Redis是一个内存数据库，也是一个常用的消息队列中间件。在Node中使用Redis作为消息队列时，可以使用Redis的发布/订阅功能来实现消息的发送和接收。

1. 发送消息：使用Redis的publish命令将消息发送到指定的通道中。

const redis = require("redis");

// 创建Redis客户端
const client = redis.createClient();

// 发送消息到指定通道
client.publish("channel", "message");

2. 接收消息：使用Redis的subscribe命令订阅指定的通道，并通过监听message事件来接收消息。

const redis = require("redis");

// 创建Redis客户端
const client = redis.createClient();

// 订阅指定通道
client.subscribe("channel");

// 监听消息事件
client.on("message", function(channel, message) {
  console.log("Received message:", message);
});

使用RabbitMQ作为消息队列

RabbitMQ是一个开源的消息队列中间件，它实现了AMQP（高级消息队列协议），提供了可靠的消息传输和强大的消息路由功能。

1. 发送消息：使用amqplib库创建一个与RabbitMQ连接的通道，并使用sendToQueue方法发送消息到指定的队列。

const amqp = require("amqplib");

async function sendMessage() {
  try {
    // 创建与RabbitMQ的连接
    const connection = await amqp.connect("amqp://localhost");
  
    // 创建通道
    const channel = await connection.createChannel();
  
    // 声明队列
    await channel.assertQueue("queue");

    // 发送消息
    channel.sendToQueue("queue", Buffer.from("message"));
  
    // 关闭连接
    await channel.close();
    await connection.close();
  } catch (error) {
    console.error("Error sending message:", error);
  }
}

sendMessage();

2. 接收消息：同样使用amqplib库创建一个与RabbitMQ连接的通道，并使用consume方法从指定的队列中接收消息。

const amqp = require("amqplib");

async function receiveMessage() {
  try {
    // 创建与RabbitMQ的连接
    const connection = await amqp.connect("amqp://localhost");
  
    // 创建通道
    const channel = await connection.createChannel();
  
    // 声明队列
    await channel.assertQueue("queue");
  
    // 接收消息
    channel.consume("queue", function(message) {
      console.log("Received message:", message.content.toString());
    }, { noAck: true });
  
    // 关闭连接
    await channel.close();
    await connection.close();
  } catch (error) {
    console.error("Error receiving message:", error);
  }
}

receiveMessage();

使用Kafka作为消息队列

Kafka是一个分布式流处理平台，也是一个常用的消息队列中间件。在Node中使用Kafka时，可以使用node-rdkafka库来进行消息的发送和接收。

1. 发送消息：使用node-rdkafka库创建一个生产者实例，并使用produce方法发送消息到指定的主题。

const Kafka = require("node-rdkafka");

const producer = new Kafka.Producer({
  "metadata.broker.list": "localhost:9092",
});

// 发送消息
producer.produce("topic", null, Buffer.from("message"));

// 关闭生产者
producer.disconnect();

2. 接收消息：使用node-rdkafka库创建一个消费者实例，并使用consume方法从指定的主题中接收消息。

const Kafka = require("node-rdkafka");

const consumer = new Kafka.KafkaConsumer({
  "group.id": "group",
  "metadata.broker.list": "localhost:9092",
});

// 订阅主题
consumer.subscribe(["topic"]);

// 监听消息事件
consumer.on("data", function(message) {
  console.log("Received message:", message.value.toString());

  // 手动提交偏移量
  consumer.commitMessageSync(message);
});

// 启动消费者
consumer.connect();

// 关闭消费者
consumer.disconnect();

c语言编程笔录