RabbitMQ中的Work Queues模式

server/2024/12/18 0:42:47/

在现代分布式系统中,消息队列(Message Queue)是实现异步通信和解耦系统的关键组件之一。RabbitMQ 是一个广泛使用的开源消息代理软件,支持多种消息传递模式。其中,Work Queues(工作队列)模式是一种常见的模式,用于在多个消费者之间分配任务,从而实现负载均衡和提高系统的处理能力。下面将详细介绍 RabbitMQ 中的 Work Queues 模式。

1. 什么是 Work Queues 模式?

Work Queues 模式(也称为任务队列模式)是一种消息传递模式,用于在多个消费者之间分配任务。在这种模式下,生产者将任务(消息)发送到队列中,多个消费者从队列中获取任务并进行处理。每个任务只会被一个消费者处理,从而实现负载均衡。

在这里插入图片描述

Work Queues 模式的主要优点包括:

  1. 负载均衡:多个消费者可以并行处理任务,从而提高系统的处理能力。
  2. 解耦:生产者和消费者之间通过队列进行通信,彼此之间不需要直接交互。
  3. 异步处理:任务可以异步处理,生产者不需要等待任务完成即可继续执行其他操作。

2. Work Queues 模式的工作原理

2.1 生产者(Producer)

生产者负责将任务(消息)发送到队列中。生产者不需要知道有多少消费者会处理这些任务,只需要将任务发送到队列即可。

2.2 队列(Queue)

队列是消息的缓冲区,用于存储生产者发送的任务。队列可以有多个消费者,但每个任务只会被一个消费者处理。

2.3 消费者(Consumer)

消费者从队列中获取任务并进行处理。多个消费者可以并行处理任务,从而实现负载均衡。消费者可以是独立的进程、线程或服务。

2.4 消息确认(Message Acknowledgment)

为了确保任务能够可靠地处理,RabbitMQ 提供了消息确认机制。消费者在处理完任务后,需要向 RabbitMQ 发送确认消息,告知任务已经处理完成。如果消费者在处理任务时崩溃,RabbitMQ 会将未确认的任务重新分配给其他消费者。

2.5 公平分发(Fair Dispatch)

默认情况下,RabbitMQ 会按顺序将任务分发给消费者。然而,如果某些消费者处理任务的速度较慢,可能会导致任务堆积。为了避免这种情况,可以使用 basicQos 方法设置预取计数(prefetch count),限制每个消费者一次可以获取的任务数量,从而实现更公平的分发。

3. 环境准备

在开始之前,确保你已经安装了以下环境:

  • Java 开发环境(JDK 8 或更高版本)
  • RabbitMQ 服务器(已启动并运行)
  • Maven(用于管理依赖)

3.1 添加依赖

首先,在你的项目中添加 RabbitMQ 客户端库的依赖。如果你使用的是 Maven,可以在 pom.xml 中添加以下依赖:

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version>
</dependency>

4. 代码案例

4.1 生产者代码

生产者负责将任务发送到队列。以下是一个简单的生产者代码示例:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {private static final String QUEUE_NAME = "work_queues";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.200.138");factory.setPort(5672);factory.setVirtualHost("/test");factory.setUsername("test");factory.setPassword("test");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, true, false, false, null);//4.将消息发送到队列for (int i = 1; i <= 10; i++) {String message = "Task to be processed, " + i;channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}}}
}

代码解释:

  • ConnectionFactory: 用于创建与 RabbitMQ 服务器的连接。
  • Connection: 表示与 RabbitMQ 服务器的物理连接。
  • Channel: 表示与 RabbitMQ 服务器的逻辑连接,用于发送和接收消息。
  • queueDeclare: 声明一个队列,true 表示队列是持久的。
  • basicPublish: 将消息发布到队列,使用默认交换机(空字符串)。

4.2 消费者代码

消费者负责从队列中获取任务并处理。以下是一个简单的消费者代码示例:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static final String QUEUE_NAME = "work_queues";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.200.138");factory.setPort(5672);factory.setVirtualHost("/test");factory.setUsername("test");factory.setPassword("test");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");channel.basicQos(1); // 每次只处理一个消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");try {doWork(message);} finally {System.out.println(" [x] Done");//手动ack,因为不同的机器处理速度不一样,因此不同的机器会在不同时间应答,这样机器就可以根据实际能力处理了channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}private static void doWork(String task) {for (char ch : task.toCharArray()) {if (ch == '.') {try {Thread.sleep(1000);} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}}}
}

代码解释:

  • basicQos(1): 设置每次只处理一个消息,确保任务的公平分配。
  • DeliverCallback: 定义消息处理逻辑,doWork 方法模拟任务处理过程。
  • basicAck: 确认消息处理完成,从队列中移除消息。

5. 运行示例

  1. 启动 RabbitMQ 服务器:确保 RabbitMQ 服务器已启动并运行。
  2. 运行多个消费者:启动多个消费者实例,确保它们连接到同一个队列。
  3. 运行生产者:启动生产者实例,发送任务到队列。

示例输出:

  • 生产者输出

     [x] Sent 'Task to be processed, 1'[x] Sent 'Task to be processed, 2'[x] Sent 'Task to be processed, 3'[x] Sent 'Task to be processed, 4'[x] Sent 'Task to be processed, 5'[x] Sent 'Task to be processed, 6'[x] Sent 'Task to be processed, 7'[x] Sent 'Task to be processed, 8'[x] Sent 'Task to be processed, 9'[x] Sent 'Task to be processed, 10'
    
  • 消费者1输出

     [*] Waiting for messages. To exit press CTRL+C[x] Received 'Task to be processed, 1'[x] Done[x] Received 'Task to be processed, 4'[x] Done[x] Received 'Task to be processed, 6'[x] Done[x] Received 'Task to be processed, 8'[x] Done[x] Received 'Task to be processed, 10'[x] Done
    
  • 消费者2输出

     [*] Waiting for messages. To exit press CTRL+C[x] Received 'Task to be processed, 2'[x] Done[x] Received 'Task to be processed, 3'[x] Done[x] Received 'Task to be processed, 5'[x] Done[x] Received 'Task to be processed, 7'[x] Done[x] Received 'Task to be processed, 9'[x] Done
    

在这里插入图片描述

6. 总结

本文详细介绍了如何在 RabbitMQ 中实现 Work Queues 模式,包括生产者、默认交换机、队列和多个消费者的设计与实现。通过使用 RabbitMQ 的 Java 客户端库,我们可以轻松地实现任务的分配和处理。Work Queues 模式非常适合需要将任务分配给多个消费者处理的场景,如任务调度、日志处理等。


http://www.ppmy.cn/server/151035.html

相关文章

【FFmpeg】FFmpeg 内存结构 ⑥ ( 搭建开发环境 | AVPacket 创建与释放代码分析 | AVPacket 内存使用注意事项 )

文章目录 一、搭建开发环境1、开发环境搭建参考2、项目搭建 二、AVPacket 创建与释放代码分析1、AVPacket 创建与释放代码2、Qt 单步调试方法3、单步调试 - 分析 AVPacket 创建与销毁代码 三、AVPacket 内存使用注意事项1、谨慎使用 av_init_packet 函数2、av_init_packet 函数…

【QT常用技术讲解】使用QMovie+QLabel播放gif动态图片,实现“正在加载”功能(源代码在资源中下载)

前言 界面在实现事件等待时&#xff0c;通过会显示一个转圈圈的动态图片&#xff0c;表示“正在加载”&#xff0c;事件完成之后关闭图片&#xff0c;QT中可以使用QMovieQLabel完成gif动态图片的播放及关闭的效果。 效果图 功能讲解 1、加载动画 void MainWindow::addloadgi…

vue3中的v-model如何自定义修饰符

子组件封装 <script setup lang"ts"> import { defineModel } from "vue"const firstName defineModel("first-name") // 自定义组件修饰符&#xff0c;首字母大写 const [lastName, lastNameModifiers] defineModel("last-name&q…

3D 生成重建036-HyperDiffusion直接生成nerf几何

3D 生成重建036-HyperDiffusion直接生成nerf几何 文章目录 0 论文工作1 论文方法2 实验结果 0 论文工作 作者提出了一种名为HyperDiffusion的新型方法&#xff0c;用于直接生成隐式神经辐射场&#xff0c;方法是预测其权重空间中的参数。与以往需要对神经网络的单个实例进行优…

FedAdam算法:供给方信用,数据质量;更新一致性

FedAdam算法:供给方信用,数据质量;更新一致性 FedAdam算法概述 FedAdam是一种联邦学习(Federated Learning)算法。联邦学习是一种机器学习技术,它允许在多个设备或数据中心(称为客户端)上训练模型,而无需将数据集中到一个中央服务器,从而保护数据隐私。FedAdam主要用于…

Python+OpenCV系列:图像的运算

文章目录 PythonOpenCV系列&#xff1a;图像的加权和、覆盖1. 图像加权和&#xff08;加权融合&#xff09;2. 图像覆盖&#xff08;区域叠加&#xff09;3. 应用场景4. 总结 PythonOpenCV系列&#xff1a;图像的加权和、覆盖 在图像处理中&#xff0c;图像的加权和与覆盖是两…

Lumos学习王佩丰Excel第二十一讲:经典Excel动态图表实现原理

一、动态图表实现原理 1、理解图表中的数据系列 在Excel图表中&#xff0c;系列指的是图表中的数据集合&#xff0c;它通常代表着一个数据源。每个系列都可以包含多个数据点&#xff0c;这些数据点在图表中以特定的形式展现&#xff0c;如柱状图中的柱子&#xff0c;折线图中…

《网络安全编程基础》之Socket编程

我的代码 server.c // server.cpp : Defines the entry point for the console application. //#include "stdafx.h" #include <Winsock2.h> #pragma comment(lib,"ws2_32.lib") //添加静态链接库文件 void main(int argc,char* argv[]) {WSADATA …