RabbitMQ简单模式和工作模式

本文介绍: RabbitMQ 是一个消息队列中间件，用于在分布式系统中进行消息传递。在 RabbitMQ 中，有几种工作模式，其中简单模式和工作模式是其中两种基本的模式之一。在 RabbitMQ 中，简单模式和工作模式的实现通常使用一些基本的概念，包括生产者、消费者、队列和消息。生产者负责发送消息到队列，而消费者则负责从队列中接收和处理消息。下面是一个使用 RabbitMQ 和 Node.js（使用amqplib库）以及 TypeScript 实现工作模式的简单示例。在这个例子中，我们将使用amqplib。

RabbitMQ 是一个消息队列中间件，用于在分布式系统中进行消息传递。在 RabbitMQ 中，有几种工作模式，其中简单模式和工作模式是其中两种基本的模式之一。

在 RabbitMQ 中，简单模式和工作模式的实现通常使用一些基本的概念，包括生产者、消费者、队列和消息。生产者负责发送消息到队列，而消费者则负责从队列中接收和处理消息。

下面是一个使用 RabbitMQ 和 Node.js（使用 amqplib 库）以及 TypeScript 实现工作模式的简单示例。在这个例子中，我们将使用 amqplib 库来连接 RabbitMQ 服务器，并使用 TypeScript 来编写代码。

首先，确保你已经安装了 amqplib 库。可以使用以下命令进行安装：

npm install amqplib

接下来，创建一个生产者和一个消费者的 TypeScript 文件。以下是示例代码：

producer.ts:

import * as amqp from 'amqplib';

async function produce() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'work_queue';

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

  for (let i = 0; i < 10; i++) {
    const message = `Message ${i}`;
    channel.sendToQueue(queue, Buffer.from(message), { persistent: true });
    console.log(` [x] Sent '${message}'`);
  }

  setTimeout(() => {
    connection.close();
    process.exit(0);
  }, 500);
}

produce();

import * as amqp from 'amqplib';

async function consume() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'work_queue';

  await channel.assertQueue(queue, { durable: true });
   // 设置每次只处理一个消息[平均分配的概念,不会让一个work太忙和太闲]
    //这一行代码的作用是告诉 RabbitMQ 不要在消费者未确认（ack）之前向其发送新的消息
  await channel.prefetch(1);

  console.log(' [*] Waiting for messages. To exit press CTRL+C');

  await channel.consume(queue, async (msg) => {
    if (msg !== null) {
      const message = msg.content.toString();
      console.log(` [x] Received ${message}`);

      // Simulate some work
      await new Promise(resolve => setTimeout(resolve, 1000));

      console.log(' [x] Done');
      channel.ack(msg);
    }
  });
}

consume();

这个示例中，生产者将消息发送到名为 work_queue 的队列中，而消费者则监听该队列并处理消息。消费者使用 channel.prefetch(1) 来确保一次只接收一个消息，从而实现竞争消费者模式。

tsc producer.ts
tsc consumer.ts

然后，分别运行 producer.js 和 consumer.js。这样你就可以在 RabbitMQ 中看到消息的生产和消费过程。

channel.consume(queueName, async (msg: Message | null) => {
  if (msg) {
    const data: EmailTask = JSON.parse(msg.content.toString());
    console.log('Processing mail task:', msg.content.toString());
    try {
      //模拟邮件发送
      await new Promise(resolve => setTimeout(resolve, 1000));
      console.log(' [x] Done');
      channel.ack(msg);
    } catch (error) {
      console.log('error:', data);
      // 处理消息失败，判断是否需要进行重试
      if (canRetry(msg)) {
        // 重新入队，进行下一次尝试
        channel.reject(msg, true);
      } else {
        // 不进行重试，将消息从队列中移除
        channel.ack(msg);
      }
    }
  }
}, { noAck: false });//默认false

await channel.consume(queueName, async (msg: Message | null) => {
  if (msg) {
    const data: EmailTask = JSON.parse(msg.content.toString());
    let retryCount = msg.properties.headers['x-retry-count'] || 0;
    console.log('Processing message:', data);
    console.log('Retry count:', retryCount);
    try {
      if (data.to.includes('recipient1@example.com')) {
        throw new Error('邮件发送失败...');
      }
      //发送邮件
      await new Promise(resolve => setTimeout(resolve, 1000));
      channel.ack(msg);
    } catch (error) {
      console.log('error:', data);
      // 增加重试次数
      retryCount++;
      // 判断是否达到最大重试次数
      if (retryCount < maxRetryAttempts) {
        // 重新发送消息到队列
        channel.sendToQueue(queueName, msg.content, {
          persistent: true,
          headers: {
            'x-retry-count': retryCount,
          },
        });
      } else {
        // 不进行重试，将消息从队列中移除
        channel.ack(msg);
      }
    }
  }
});