RabbitMQ中的Work Queues模式
在现代分布式系统中,消息队列(Message Queue)是实现异步通信和解耦系统的关键组件之一。RabbitMQ 是一个广泛使用的开源消息代理软件,支持多种消息传递模式。其中,Work Queues(工作队列)模式是一种常见的模式,用于在多个消费者之间分配任务,从而实现负载均衡和提高系统的处理能力。下面将详细介绍 RabbitMQ 中的 Work Queues 模式。
1. 什么是 Work Queues 模式?
Work Queues 模式(也称为任务队列模式)是一种消息传递模式,用于在多个消费者之间分配任务。在这种模式下,生产者将任务(消息)发送到队列中,多个消费者从队列中获取任务并进行处理。每个任务只会被一个消费者处理,从而实现负载均衡。
Work Queues 模式的主要优点包括:
- 负载均衡:多个消费者可以并行处理任务,从而提高系统的处理能力。
- 解耦:生产者和消费者之间通过队列进行通信,彼此之间不需要直接交互。
- 异步处理:任务可以异步处理,生产者不需要等待任务完成即可继续执行其他操作。
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. 运行示例
- 启动 RabbitMQ 服务器:确保 RabbitMQ 服务器已启动并运行。
- 运行多个消费者:启动多个消费者实例,确保它们连接到同一个队列。
- 运行生产者:启动生产者实例,发送任务到队列。
示例输出:
-
生产者输出:
[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 模式非常适合需要将任务分配给多个消费者处理的场景,如任务调度、日志处理等。
相关文章:

RabbitMQ中的Work Queues模式
在现代分布式系统中,消息队列(Message Queue)是实现异步通信和解耦系统的关键组件之一。RabbitMQ 是一个广泛使用的开源消息代理软件,支持多种消息传递模式。其中,Work Queues(工作队列)模式是一…...

GESP202412 四级【Recamán】题解(AC)
》》》点我查看「视频」详解》》》 [GESP202412 四级] Recamn 题目描述 小杨最近发现了有趣的 Recamn 数列,这个数列是这样生成的: 数列的第一项 a 1 a_1 a1 是 1 1 1;如果 a k − 1 − k a_{k-1}-k ak−1−k 是正整数并且没有在数…...

蓝桥杯新年题解 | 第15届蓝桥杯迎新篇
蓝桥杯新年题解 | 第15届蓝桥杯迎新篇 2024年的蓝桥杯即将拉开序幕!对于许多编程爱好者来说,这不仅是一次展示自我能力的舞台,更是一次学习和成长的机会。作为一名大一新生的小蓝,对蓝桥杯充满了期待,但面对初次参赛的…...

3D 生成重建035-DiffRF直接生成nerf
3D 生成重建035-DiffRF直接生成nerf 文章目录 0 论文工作1 论文方法2 实验结果 0 论文工作 本文提出了一种基于渲染引导的三维辐射场扩散新方法DiffRF,用于高质量的三维辐射场合成。现有的方法通常难以生成具有细致纹理和几何细节的三维模型,并且容易出…...
@SpringBootTest 报错: UnsatisfiedDependencyException
Spring Boot Test 报错: UnsatisfiedDependencyException 在使用 SpringBootTest 测试时,出现 UnsatisfiedDependencyException 报错,原因和解决方法如下。 报错原因分析 1. Spring 存在涉及 Bean 没有被添加 Spring Boot 测试中,默认会加…...
mysql、postgresql、oceanbase调优
一、mysql 1、my.cnf [mysqld_safe] log-error=/data/mysql/log/mysql.log pid-file=/data/mysql/run/mysqld.pid[client] socket=/data/mysql/run/mysql.sock default-character-set=utf8[mysqld] basedir=/usr/local/mysql tmpdir=/data/mysql/tmp datadir=/data/mysql/dat…...
MySQL 数据库事务实践
引言 在现代应用程序开发中,确保数据库操作的完整性和一致性至关重要。MySQL 提供了强大的事务管理功能,允许开发者以原子性、一致性、隔离性和持久性(ACID)的方式处理数据。本文将通过详细的解释和实际示例,带你深入…...

VScode、Windsurf、Cursor 中 R 语言相关快捷键设置
前言 在生物信息学数据分析中,R语言是一个不可或缺的工具。为了提高R语言编程效率,合理设置快捷键显得尤为重要。本文介绍在VSCode Windsurf Cursor 中一些实用的R语言快捷键设置,让非 Rstudio 的 IDE 用起来得心应手😑 操作种…...
tcpdump编译
https://github.com/westes/flex/releases/download/v2.6.4/flex-2.6.4.tar.gz tar -zxvf flex-2.6.4.tar.gz ./configure CFLAGS-D_GNU_SOURCE make sudo make installwget http://ftp.gnu.org/gnu/bison/bison-3.2.1.tar.gz ./configure make sudo make install以上两个库是…...
Linux下禁止root远程登录访问
开始讲故事 Long long ago, Linux远程访问方式有telnet、ssh两种协议;有人可能还会说vnc和rdp协议方式,后面这两种主要是可视化桌面场景下的,并非主流。 时过境迁,telnet因安全性低逐渐被禁用淘汰,最后就…...

算法刷题Day16: BM41 输出二叉树的右视图
题目链接 描述 思路: 递归构造二叉树在Day15有讲到。复习一下,就是使用递归构建左右子树。将中序和前序一分为二。 接下来是找出每一层的最右边的节点,可以利用队列层次遍历。 利用队列长度记录当前层有多少个节点,每次从队列里…...

登录授权的实现:json web token + redis + springboot
文章目录 引言I token实现思路传统JWT TOKEN认证方式改进的JWT TOKEN认证方式redis设计II java代码实现登录接口退出登录接口登录之后接口(token解析和校验)III 常见问题400引言 应用场景: 登录认证 I token实现思路 传统JWT TOKEN认证方式 RESTful API TOKEN认证方式:…...

yolov,coco,voc标记的睡岗检测数据集,可识别在桌子上趴着睡,埋头睡觉,座椅上靠着睡,平躺着睡等多种睡姿的检测,6549张图片
yolov,coco,voc标记的睡岗检测数据集,可识别在桌子上趴着睡,埋头睡觉,座椅上靠着睡,平躺着睡等多种睡姿的检测,6549张图片 数据集分割 6549总图像数 训练组91% 5949图片 有效集9&#x…...
数据库表的CRUD
SQL语句(Structured Query Language)是用于与关系型数据库进行交互的语言。下面是几个常用的SQL语句: 创建表: CREATE TABLE table_name ( column1 datatype, column2 datatype, column3 datatype, ... ); 插入数据: …...

Proxy与Reflect
监听对象操作 在Object中,可以通过defineProperty中的get,set进行监听, Proxy基本使用 有两个参数,第一个是要代理的对象,第二个是捕获器,在不知道捕获器使用哪个之前可以先传个空对象。就会启用默认的捕获…...

【安卓开发】【Android Studio】启动时报错“Unable to access Android SDK add-on list”
一、问题描述 在启动Android Studio时,软件报错:Unable to access Android SDK add-on list,报错截图如下: 二、原因及解决方法 初步推测是由于网络节点延迟,无法接入谷歌导致的。点击Cancel取消即可。...

【C语言篇】C 语言总复习(下):点亮编程思维,穿越代码的浩瀚星河
我的个人主页 我的专栏:C语言,希望能帮助到大家!!!点赞❤ 收藏❤ 在C语言的世界里,结构体和联合体以及文件操作都是非常重要且实用的知识板块,掌握它们能帮助我们更高效地组织数据以及与外部文…...

AI技术架构:从基础设施到应用
人工智能(AI)的发展,正以前所未有的速度重塑我们的世界。了解AI技术架构,不仅能帮助我们看懂 AI 的底层逻辑,还能掌握其对各行业变革的潜力与方向。 一、基础设施层:AI 技术的坚实地基 基础设施层是 AI 技…...

centos7的yum镜像源设置
sudo yum repolist 查看镜像源连接情况,not found即为连接失败 sudo cp -r /etc/yum.repos.d /etc/yum.repos.d.backup 备份镜像源文件 sudo nano /etc/yum.repos.d/CentOS-Base.repo 进入镜像源文件编辑内容 # CentOS-Base.repo # # The mirror system uses the…...

Qt6开发自签名证书的https代理服务器
目标:制作一个具备类似Fiddler、Burpsuit、Wireshark的https协议代理抓包功能,但是集成到自己的app内,这样无需修改系统代理设置,使用QWebengineview通过自建的代理服务器,即可实现https包的实时监测、注入等自定义功能…...

wordpress后台更新后 前端没变化的解决方法
使用siteground主机的wordpress网站,会出现更新了网站内容和修改了php模板文件、js文件、css文件、图片文件后,网站没有变化的情况。 不熟悉siteground主机的新手,遇到这个问题,就很抓狂,明明是哪都没操作错误&#x…...

龙虎榜——20250610
上证指数放量收阴线,个股多数下跌,盘中受消息影响大幅波动。 深证指数放量收阴线形成顶分型,指数短线有调整的需求,大概需要一两天。 2025年6月10日龙虎榜行业方向分析 1. 金融科技 代表标的:御银股份、雄帝科技 驱动…...
Android Wi-Fi 连接失败日志分析
1. Android wifi 关键日志总结 (1) Wi-Fi 断开 (CTRL-EVENT-DISCONNECTED reason3) 日志相关部分: 06-05 10:48:40.987 943 943 I wpa_supplicant: wlan0: CTRL-EVENT-DISCONNECTED bssid44:9b:c1:57:a8:90 reason3 locally_generated1解析: CTR…...
基于数字孪生的水厂可视化平台建设:架构与实践
分享大纲: 1、数字孪生水厂可视化平台建设背景 2、数字孪生水厂可视化平台建设架构 3、数字孪生水厂可视化平台建设成效 近几年,数字孪生水厂的建设开展的如火如荼。作为提升水厂管理效率、优化资源的调度手段,基于数字孪生的水厂可视化平台的…...
Qt Http Server模块功能及架构
Qt Http Server 是 Qt 6.0 中引入的一个新模块,它提供了一个轻量级的 HTTP 服务器实现,主要用于构建基于 HTTP 的应用程序和服务。 功能介绍: 主要功能 HTTP服务器功能: 支持 HTTP/1.1 协议 简单的请求/响应处理模型 支持 GET…...
Caliper 配置文件解析:config.yaml
Caliper 是一个区块链性能基准测试工具,用于评估不同区块链平台的性能。下面我将详细解释你提供的 fisco-bcos.json 文件结构,并说明它与 config.yaml 文件的关系。 fisco-bcos.json 文件解析 这个文件是针对 FISCO-BCOS 区块链网络的 Caliper 配置文件,主要包含以下几个部…...
什么?连接服务器也能可视化显示界面?:基于X11 Forwarding + CentOS + MobaXterm实战指南
文章目录 什么是X11?环境准备实战步骤1️⃣ 服务器端配置(CentOS)2️⃣ 客户端配置(MobaXterm)3️⃣ 验证X11 Forwarding4️⃣ 运行自定义GUI程序(Python示例)5️⃣ 成功效果
注:当前使用的是 ol 5.3.0 版本,天地图使用的key请到天地图官网申请,并替换为自己的key 地图分屏对比在WebGIS开发中是很常见的功能,和卷帘图层不一样的是,分屏对比是在各个地图中添加相同或者不同的图层进行对比查看。…...

通过MicroSip配置自己的freeswitch服务器进行调试记录
之前用docker安装的freeswitch的,启动是正常的, 但用下面的Microsip连接不上 主要原因有可能一下几个 1、通过下面命令可以看 [rootlocalhost default]# docker exec -it freeswitch fs_cli -x "sofia status profile internal"Name …...
绕过 Xcode?使用 Appuploader和主流工具实现 iOS 上架自动化
iOS 应用的发布流程一直是开发链路中最“苹果味”的环节:强依赖 Xcode、必须使用 macOS、各种证书和描述文件配置……对很多跨平台开发者来说,这一套流程并不友好。 特别是当你的项目主要在 Windows 或 Linux 下开发(例如 Flutter、React Na…...