完整指南：使用 Nodejs 进行消息传递-js教程-PHP中文网

完整指南：使用 Nodejs 进行消息传递

心靈之曲

发布： 2024-11-10 18:27:01

转载

707人浏览过

完整指南：使用 nodejs 进行消息传递

node.js 中的消息传递是创建可扩展、弹性和异步系统的基本实践，尤其是在基于微服务的架构中。本指南涵盖了 rabbitmq 和 kafka 等流行库的实际实施的基本概念。

1.什么是消息传递以及为什么使用它？

消息传递是在服务或软件组件之间发送、接收和管理消息的过程。它适用于：

解耦：允许服务独立。
可扩展性：通过分发消息来管理高流量负载。
弹性：即使在出现临时故障的情况下也能确保消息处理。

常见使用场景:

2.配置 node.js 环境

安装 node.js：确保您安装了最新版本的 node.js。
包管理器：使用npm或yarn安装依赖项。
基本依赖：

npm install dotenv amqplib

登录后复制

3.消息传递协议和工具

rabbitmq：

rabbitmq 是一种广泛使用的 amqp 代理，用于交换消息。

阿帕奇卡夫卡：

非常适合大规模数据流。

其他选项：

redis

4.使用 rabbitmq 的基本实现

第 1 步：配置 rabbitmq 服务器

docker

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

登录后复制

第 2 步：连接到 rabbitmq

使用 amqplib 库创建连接和队列。

const amqp = require('amqplib');

async function connect() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createchannel();
    const queue = 'tasks';

    await channel.assertqueue(queue, { durable: true });

    console.log(`waiting for messages in ${queue}`);
    channel.consume(queue, (msg) => {
      console.log(`received: ${msg.content.tostring()}`);
      channel.ack(msg);
    });
  } catch (err) {
    console.error('error:', err);
  }
}

connect();

登录后复制

5.使用 kafka 实现

kafka 需要 kafkajs 库。

初始设置

   npm install kafkajs

登录后复制

使用 kafkajs 的生产者和消费者

制作人：

const { kafka } = require('kafkajs');

const kafka = new kafka({ clientid: 'my-app', brokers: ['localhost:9092'] });

const producer = kafka.producer();

async function sendmessage() {
  await producer.connect();
  await producer.send({
    topic: 'test-topic',
    messages: [{ value: 'hello kafkajs' }],
  });
  await producer.disconnect();
}

sendmessage();

登录后复制

消费者：

const consumer = kafka.consumer({ groupId: 'test-group' });

async function consumeMessages() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        partition,
        value: message.value.toString(),
      });
    },
  });
}

consumeMessages();

登录后复制