当前位置： 首页 > news >正文

RabbitMQ简单模式和工作模式

news 2025/6/21 17:07:21

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

  1. 简单模式（Simple Mode）：

    • 在简单模式中，有一个生产者（Producer）将消息发送到一个队列（Queue）中，然后有一个消费者（Consumer）从队列中接收并处理消息。
    • 这是最基本的消息队列模式，适用于单个生产者和单个消费者的场景。
    • 生产者将消息发送到队列，而消费者从队列中接收并处理消息，消息的传递是单向的。

  2. 工作模式（Work Queue Mode）：

    • 工作模式也被称为竞争消费者模式。在这种模式下，有多个消费者监听同一个队列，但每条消息只能被其中一个消费者接收和处理。
    • 当消息被发送到队列时，它将被发送给下一个空闲的消费者，从而实现消息的分发和并发处理。
    • 这种模式对于处理大量工作的情况很有用，可以通过增加消费者的数量来提高消息处理的速度。

在 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();

consumer.ts:

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 workawait new Promise(resolve => setTimeout(resolve, 1000));console.log(' [x] Done');channel.ack(msg);}});
}consume();

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

记得在运行前启动 RabbitMQ 服务器，并确保 TypeScript 文件已编译成 JavaScript。你可以使用以下命令进行编译：

tsc producer.ts
tsc consumer.ts

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

RabbitMQ消息持久化和手动应答

在 RabbitMQ 中，消息持久化和手动应答是两个关键的概念，它们可以帮助确保消息的可靠传递和处理。下面简要介绍这两个概念：

  1. 消息持久化（Message Durability）：

    • 默认情况下，RabbitMQ 中的消息是瞬时的，也就是说，如果 RabbitMQ 服务器停止或崩溃，所有未处理的消息都会丢失。
    • 通过将消息标记为持久化，你可以确保消息在 RabbitMQ 服务器重启后仍然可用。要实现消息持久化，需要在发送消息时设置消息的 deliveryMode 属性为 2（persistent）。
    • 例如，在生产者端设置消息为持久化：
    channel.sendToQueue(queue, Buffer.from(message), { persistent: true });
    • 在消费者端，你需要确保队列和消息都被声明为持久化：
    channel.assertQueue(queue, { durable: true });

    这样，即使 RabbitMQ 服务器重启，持久化的消息也不会丢失。

  2. 手动应答（Manual Acknowledgment）：

    • 默认情况下，RabbitMQ 使用自动应答（auto-acknowledgment）模式，即一旦消息被传递给消费者，RabbitMQ 就将其标记为已处理。
    • 在某些情况下，你可能需要更细粒度的控制，以确保消息在被消费者完全处理之后才被标记为已处理。这就是手动应答的用途。
    • 在消费者端，需要将 noAck 设置为 false，表示手动应答模式：
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
  • 在这种情况下，消费者需要在处理完消息后显式调用 channel.ack(msg) 来确认消息已被处理。如果消费者崩溃或在处理消息时发生错误，消息将保持在队列中，直到被明确确认。
  • 在 RabbitMQ 中，channel.reject 方法用于拒绝一条消息。它的参数如下channel.reject(msg, requeue);
    msg: 要拒绝的消息对象。
    requeue: 如果设置为 true，则被拒绝的消息将被重新排队，即重新放回队列。如果设置为 false，则消息将被删除。默认为 true。

综合使用消息持久化和手动应答，可以确保在面对不同情况时，消息的可靠传递和处理。

重试间隔和次数

在这里插入图片描述
在这里插入图片描述

  1. 重新投递消息并设置头部信息：

    • 在处理消息失败时，将消息重新投递到队列，并设置一个头部信息，例如 x-redelivered-count，用来记录消息的重试次数。
    • 在消费者端，根据这个头部信息来判断是否达到重试次数的上限，如果是，则不再重新投递，可能将消息放入死信交换机。

  2. 使用外部存储记录重试次数：

    • 每次消息处理失败时，将消息的唯一标识（例如 UUID）和重试次数记录到外部存储中（例如 Redis、Memcache、MySQL）。
    • 在消费者端，在每次重新处理时，从外部存储中获取当前重试次数，并判断是否达到重试次数的上限。

  3. 自定义插件：

    • 编写一个 RabbitMQ 插件，实现自定义的消息重试逻辑，包括记录重试次数、判断是否重新投递等。
    • 这样可以更灵活地控制消息的处理流程。

需要注意的是，这些方法都是基于 RabbitMQ 不直接提供重试次数限制的情况下的一些自定义实践。在回答中也提到了关于 quorum queues 的更新，以及支持通过策略（policy）来限制重投递次数的可能性。因此，具体的实现方式可能会随着 RabbitMQ 版本的更新而有所变化。

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);}}}
});

请添加图片描述

http://www.lryc.cn/news/289905.html

相关文章：

  • c语言实战之贪吃蛇
  • Midjourney图片生成描述词记录（今天一天）
  • 类和对象 第五部分第四小节：赋值运算符重载
  • Django从入门到精通（一）
  • 数据库分表分库的原则
  • Java技术栈 —— Docker容器
  • Mysql大数据量分页优化
  • QT tcp与udp网络通信以及定时器的使用 (7)
  • web架构师编辑器内容-添加自动保存的功能
  • 【Redis】关于它为什么快？使用场景？以及使用方式？为何引入多线程？
  • SpringBoot之JWT登录
  • 【备战蓝桥杯】——循环结构
  • 【数据分享】1929-2023年全球站点的逐年平均气温数据（Shp\Excel\免费获取）
  • 探索Pyecharts关系图绘制技巧：炫酷效果与创意呈现【第42篇—python：Pyecharts水球图】
  • 蓝桥杯-循环节长度
  • Jython调用openwire库连接activemq转发topic订阅消息到另一个activemq 服务器上 完整代码
  • 面试经典题---30.串联所有单词的子串
  • 字符串随机生成工具(开源)-Kimen(奇门)
  • UE4 CustomDepthMobile流程小记
  • Docker 基础篇
  • Idea上操作Git回退本地版本，怎么样保留已修改的文件，回退本地版本的四种方式代表什么?
  • vue3封装el-pagination分页组件
  • 负载均衡下Webshell连接思路及难点
  • 基于链表实现贪吃蛇游戏
  • Python网络爬虫实战——实验6：Python实现js逆向与加解密
  • 【python】使用aiohttp库编写一个简单的异步服务器
  • 新手使用代理IP接入代码教程
  • JVM问题排查手册
  • 前端canvas项目实战——简历制作网站(三)——右侧属性栏(线条宽度样式)
  • 字节跳动二面经典题目