Node.js 中的 Pub/Sub

2025 年 3 月 1 日 | 阅读 4 分钟

Pub/Sub 简介

在发布/订阅 (pub/sub) 通信范例中，发送者（发布者）不会直接将消息发送给特定的接收者（订阅者）。相反，它们会将消息发布到一个频道，订阅者会收到发布的通知。因此，通信模式变得更加灵活和可扩展。消息的创建和消费是分离的。这种方法在实时应用程序、微服务架构和分布式系统中非常有用。

Node.js 中的 Pub/Sub

由于 Node.js 的事件驱动、非阻塞 I/O 方法，Pub/Sub 范例非常适合 Node.js。Redis、MQTT 原生 Node.js 模块以及 RabbitMQ 和 Google Cloud Pub/Sub 等消息代理是使 Node.js 的 Pub/Sub 更易于使用的框架和技术。Node.js 的 EventEmitter 原生集成在 Node.js 框架中，通过 events 模块实现。它可以用来实现一个简单的 Pub/Sub 方法。这是一个简单的例子

 
const EventEmitter = require('events');
const ee = new EventEmitter();

// Sub 1
ee.on('msg', (m) => {
    console.log(`Sub 1 received: ${m}`);
});

// Sub 2
ee.on('msg', (m) => {
    console.log(`Sub 2 received: ${m}`);
});

// Pub
ee.emit('msg', 'Hello, World!');   

输出

说明

在这个例子中，有两个订阅者正在等待 "message" 事件的发生。发布者发送一个 "message" 事件，两个订阅者都会处理并接收到它。

Redis 发布/订阅 (Pub/Sub) 消息由内存数据结构存储 Redis 提供支持。由于其有效性和易用性，它是 Pub/Sub 部署的一个流行选择。

在 Node.js 中，如果没有 redis 包，则无法使用Redis Pub/Sub。使用 npm install redis 安装。Redis Pub/Sub 示例如下：const redis = require('redis');

 
// Create Redis clients
const pub = redis.createClient();
const sub = redis.createClient();
// Subscriber
sub.on('message', (ch, msg) => {
    console.log(`Received message: ${msg} from channel: ${ch}`);
});
sub.subscribe('my_channel');
// Publisher
pub.publish('my_channel', 'Hello, Redis!');   

输出

在您的示例中，发布者将消息发送到订阅者正在监听的同一频道“my_channel”。收到消息后，订阅者会将其记录下来。

MQTT 发布/订阅

MQTT (Message Queuing Telemetry Transport) 是一种针对物联网和低带宽环境的轻量级消息协议。它为 Node.js 提供了许多实现，并且在 Pub/Sub 方面表现良好。

如果您想在 Node.js 中使用 MQTT，则需要 MQTT 包。

使用 npm 安装 MQTT

这是一个 MQTT 发布/订阅示例

 
const mqtt = require('mqtt');
const client = mqtt.connect('mqtt://broker.hivemq.com');
// Subscriber
client.on('connect', () => {
    client.subscribe('my_topic', (err) => {
        if (!err) {
            console.log('Subscribed to my_topic');
        }
    });
});

client.on('message', (topic, message) => {
    console.log(`Received message: ${message.toString()} from topic: ${topic}`);
});
// Publisher
setTimeout(() => {
    client.publish('my_topic', 'Hello, MQTT!');
}, 1000);   

输出

说明

在此示例中，订阅者连接到 MQTT 代理并订阅 'my_topic'。发布者将消息发送到同一主题，订阅者接收并记录下来。

RabbitMQ 发布/订阅

RabbitMQ 是一个强大的消息代理，通过其交换器和队列模型支持发布/订阅。RabbitMQ 的灵活性和可靠性使其成为企业应用程序中实现发布/订阅的热门选择。

要在 Node.js 中使用 RabbitMQ，您需要 amqplib 包。

这是一个 RabbitMQ 发布/订阅示例

 
const amqp = require('amqplib/callback_api');
// Pub
amqp.connect('amqp://', (err, conn) => {
    conn.createChannel((err, ch) => {
        const ex = 'logs';
        const msg = 'Hello, RabbitMQ!';
        ch.assertExchange(ex, 'fanout', { durable: false });
        ch.publish(ex, '', Buffer.from(msg));
        console.log(`Sent: ${msg}`);
    });
    setTimeout(() => {
        conn.close();
        process.exit(0);
    }, 500);
});

// Sub
amqp.connect('amqp://', (err, conn) => {
    conn.createChannel((err, ch) => {
        const ex = 'logs';
        ch.assertExchange(ex, 'fanout', { durable: false });
        ch.assertQueue('', { exclusive: true }, (err, q) => {
            console.log(`Waiting for messages in ${q.queue}`);
            ch.bindQueue(q.queue, ex, '');

            ch.consume(q.queue, (msg) => {
                if (msg.content) {
                    console.log(`Received: ${msg.content.toString()}`);
                }
            }, { noAck: true });
        });
    });
});   

输出

订阅者启动并等待消息。它会记录它正在等待消息的队列名称。

<unique_id> 将是 RabbitMQ 为临时队列生成的唯一标识符。

发布者发送消息并记录下来。

不久之后，订阅者会收到消息并记录下来。

在此实例中，消息由发布者发送到名为“logs”的交换器，消息由订阅者从该交换器接收。'fanout' 交换器类型确保消息能到达每个订阅者。

Google Cloud Pub/Sub

您可以使用 GoogleCloud Pub/Sub 来大规模部署 Pub/Sub。这是一种完全托管的消息服务，旨在优先考虑全球分发和可靠性。

要使用 Node.js 中的 Google Cloud Pub/Sub，需要 @google-cloud/pubsub 包。

这是一个 Google Cloud Pub/Sub 示例

 
const { PubSub } = require('@google-cloud/pubsub');
const pubsub = new PubSub();

// Publisher
async function publishMessage() {
    const topicName = 'my-topic';
    const data = Buffer.from('Hello, Google Cloud Pub/Sub!');
    await pubsub.topic(topicName).publish(data);
    console.log('Message published.');
}

// Subscriber
async function listenForMessages() {
    const subscriptionName = 'my-subscription';
    const subscription = pubsub.subscription(subscriptionName);

    subscription.on('message', message => {
        console.log(`Received message: ${message.data.toString()}`);
        message.ack();
    });
}
publishMessage();
listenForMessages();   

输出

说明

在此实例中，订阅者侦听一个主题上的订阅，发布者已向该主题发送了一条消息。一旦消息发布，订阅者就会收到并处理它。

下一主题Soap-services-in-nodejs

Node.js 中的 Pub/Sub

Pub/Sub 简介