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

RabbitMQ入门实战

RabbitMQ 是一个开源的消息中间件,实现了高级消息队列协议(AMQP),用于在分布式系统中进行消息传递。它能够在应用之间传递消息,解耦应用组件,提高系统的可伸缩性和可维护性。RabbitMQ 使用高级消息队列协议(AMQP),这是一种开放的、标准化的协议,定义了消息格式、交换方式、队列管理等规范。拥有强大的社区支持,提供了广泛的文档和示例。它还支持插件机制,可以根据实际需求进行扩展。下面就对rabbitMQ进行实战。

官网:RabbitMQ: easy to use, flexible messaging and streaming — RabbitMQ

f6b6cbd4487045e1b2361c7d26bfbe21.png 

1. 基本概念

  • 消息队列:消息队列是一种在不同应用之间传递数据的机制。在RabbitMQ中,消息队列是通过Exchange(交换机)来进行消息路由的。
  • Exchange:Exchange负责将消息路由到一个或多个队列。RabbitMQ支持不同类型的Exchange,包括Direct、Fanout、Topic等。
  • Queue:Queue是消息的容器,消息在发送到Exchange后通过Routing Key被路由到一个或多个Queue中。

2. 使用场景

消息中间件经常被用来处理异步、削峰填谷,和多个组件之间进行解耦的作用。

  • 异步任务处理:RabbitMQ可以作为任务队列,将任务发布到队列中,然后由后台工作者异步处理。这在分布式系统中很常见,可以提高系统的响应速度。
  • 事件驱动架构:RabbitMQ可以用于实现事件驱动的架构,不同组件之间通过消息进行通信。当某个事件发生时,可以将消息发送到队列中,由其他组件进行消费。
  • 日志收集:RabbitMQ可以用于日志收集系统,应用程序将日志消息发送到队列中,日志收集器订阅队列并将日志保存到数据库或其他存储中。

3. RabbitMQ实战

3.1 rabbitMQ安装

这里演示使用docker-compose方式安装,创建一个docker-compose.yml文件并写入以下内容

version: '3'services:rabbitmq:image: "rabbitmq:management"container_name: "rabbitmq-container"ports:- "5672:5672"    # RabbitMQ 默认端口- "15672:15672"  # RabbitMQ 管理界面端口volumes:- "./rabbitmq-data:/var/lib/rabbitmq"  # 数据文件挂载environment:RABBITMQ_DEFAULT_USER: "rabbit"RABBITMQ_DEFAULT_PASS: "rabbit1qz"

执行命令拉取并启动容器

docker-compose up -d

80fdea88312c467aa1a0822d52a47e8c.png 执行命令查看docker容器是否正常

docker ps | grep rabbit

03ec8572cb2a4f09958874359a523ef3.png

 然后地址栏http://localhost:15672/访问rabbitMQ管理界面

4d6ab83fd04740a0bcfeaf9defb407e5.png输入用户名密码可以看到如下界面,在这个页面上可以创建Exchanges和Queue,这里就不赘述了,大家想了解的可以参考官方文档。

e4a6faaa568445d1b45a20adc2c2fc5b.png 

3.1 代码集成

上面安装完MQ组件之后,就可以用java代码进行连接测试了。使用Maven添加RabbitMQ的Java客户端库到项目里:

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.0</version> <!-- 替换为最新版本 -->
</dependency>

 

3.2 发送消息

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Channel;public class MessageSender {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost"); // RabbitMQ服务器地址// 创建连接try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 发送消息String message = "Hello, RabbitMQ!";channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}}
}

 

3.3 接收消息

import com.rabbitmq.client.*;public class MessageReceiver {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost"); // RabbitMQ服务器地址// 创建连接try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 定义消息处理器DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};// 监听队列,接收消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});// 持续监听队列,不会退出System.out.println(" [*] Waiting for messages. To exit press Ctrl+C");Thread.sleep(Long.MAX_VALUE);}}
}

 

以上示例中,MessageSender类用于发送消息到名为"hello"的队列,而MessageReceiver类用于监听该队列并接收消息。

4. RabbitMQ高级特性

RabbitMQ 提供了许多高级特性,包括持久化、消息确认、事务、死信队列等。下面将结合 Java 完整代码进行说明这些高级特性。

1. 持久化

持久化确保在 RabbitMQ 服务器重启时,队列和消息不会丢失。

代码示例

// 发布者代码
public class DurableProducer {// ...初始化 RabbitMQ 连接等代码...public void publishPersistentMessage(String message) {channel.basicPublish("", "durable_queue", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}
}// 订阅者代码
public class DurableSubscriber {// ...初始化 RabbitMQ 连接等代码...public void subscribeToPersistentMessages() {channel.queueDeclare("durable_queue", true, false, false, null);channel.basicConsume("durable_queue", true, (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 处理消息的逻辑...}, consumerTag -> {});}
}

 

2. 消息确认

消息确认确保消息已经被消费者成功处理。

代码示例

// 发布者代码
public class AckProducer {// ...初始化 RabbitMQ 连接等代码...public void publishAckMessage(String message) {channel.basicPublish("", "ack_queue", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}
}// 订阅者代码
public class AckSubscriber {// ...初始化 RabbitMQ 连接等代码...public void subscribeToAckMessages() {channel.queueDeclare("ack_queue", false, false, false, null);channel.basicConsume("ack_queue", false, (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 处理消息的逻辑...// 手动发送消息确认channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}, consumerTag -> {});}
}

 

3. 事务

RabbitMQ 支持事务,但由于性能问题,通常建议使用消息确认代替。

代码示例

// 发布者代码
public class TransactionalProducer {// ...初始化 RabbitMQ 连接等代码...public void publishTransactionalMessage(String message) throws IOException {try {channel.txSelect(); // 开启事务channel.basicPublish("", "transactional_queue", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");channel.txCommit(); // 提交事务} catch (IOException e) {channel.txRollback(); // 回滚事务e.printStackTrace();}}
}// 订阅者代码
public class TransactionalSubscriber {// ...初始化 RabbitMQ 连接等代码...public void subscribeToTransactionalMessages() {try {channel.queueDeclare("transactional_queue", false, false, false, null);while (true) {channel.txSelect(); // 开启事务GetResponse response = channel.basicGet("transactional_queue", true);if (response != null) {String message = new String(response.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 处理消息的逻辑...channel.txCommit(); // 提交事务} else {channel.txRollback(); // 回滚事务}}} catch (IOException e) {e.printStackTrace();}}
}

 

4. 死信队列

死信队列用于处理无法被消费者成功处理的消息。

代码示例

// 发布者代码
public class DeadLetterProducer {// ...初始化 RabbitMQ 连接等代码...public void publishDeadLetterMessage(String message) {Map<String, Object> headers = new HashMap<>();headers.put("x-dead-letter-exchange", "dead_letter_exchange");headers.put("x-dead-letter-routing-key", "dl_queue");AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().headers(headers).build();channel.basicPublish("", "original_queue", properties, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}
}// 订阅者代码
public class DeadLetterSubscriber {// ...初始化 RabbitMQ 连接等代码...public void subscribeToDeadLetterMessages() {channel.exchangeDeclare("dead_letter_exchange", BuiltinExchangeType.DIRECT);channel.queueDeclare("dl_queue", false, false, false, null);channel.queueBind("dl_queue", "dead_letter_exchange", "");channel.queueDeclare("original_queue", false, false, false, null);channel.queueBind("original_queue", "", "original_queue");channel.basicConsume("original_queue", false, (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 处理消息的逻辑...// 模拟处理失败,将消息发送到死信队列channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);}, consumerTag -> {});}
}

 

这些是 RabbitMQ 的一些高级特性的简单示例。在实际项目中,具体的实现可能会更加复杂,并需要根据场景进行适当的调整。

5. 总结

RabbitMQ作为一款强大的消息中间件,在异步任务处理、事件驱动架构以及日志收集等场景中都有广泛的应用。通过简单的代码实例,我们了解了RabbitMQ的基本概念以及如何在Java中使用RabbitMQ进行消息的发送和接收。

希望本文能够帮助大家入门RabbitMQ,并在实际项目中灵活应用消息队列的机制。

 

相关文章:

RabbitMQ入门实战

RabbitMQ 是一个开源的消息中间件&#xff0c;实现了高级消息队列协议&#xff08;AMQP&#xff09;&#xff0c;用于在分布式系统中进行消息传递。它能够在应用之间传递消息&#xff0c;解耦应用组件&#xff0c;提高系统的可伸缩性和可维护性。RabbitMQ 使用高级消息队列协议…...

vue3-模版引用ref

1. 介绍 概念&#xff1a;通过 ref标识 获取真实的 dom对象或者组件实例对象 2. 基本使用 实现步骤&#xff1a; 调用ref函数生成一个ref对象 通过ref标识绑定ref对象到标签 代码如下&#xff1a; 父组件&#xff1a; <script setup> import { onMounted, ref } …...

C# 十大排序算法

以下是常见的十大排序算法&#xff08;按照学习和实现的顺序排列&#xff09;&#xff1a; 冒泡排序&#xff08;Bubble Sort&#xff09;选择排序&#xff08;Selection Sort&#xff09;插入排序&#xff08;Insertion Sort&#xff09;希尔排序&#xff08;Shell Sort&…...

面试之Glide如何绑定Activity的生命周期

Glide绑定Activity生命周期 Glide.with() 下面都是它的重载方法&#xff0c;Context&#xff0c;Activity&#xff0c;FragmentActivity, Fragment, android.app.Fragment fragment,View都可以作为他的参数&#xff0c;内容大同小异&#xff0c;都是先getRetriever&#xff0…...

从 fatal 错误到 sync.Map:Go中 Map 的并发策略

为什么 Go 语言在多个 goroutine 同时访问和修改同一个 map 时&#xff0c;会报出 fatal 错误而不是 panic&#xff1f;我们该如何应对 map 的数据竞争问题呢&#xff1f; 这篇文章将带你一步步了解背后的原理&#xff0c;并引出解决 map 并发问题的方案。 Map 数据竞争 首先…...

Simon算法详解

0.0 Intro 相关的算法&#xff1a; Deutsh-Jozsa算法&#xff1a; 第一个量子算法对经典算法取得指数级加速的算法 美中不足在于只能确定函数是平衡的还是非平衡的&#xff0c;无法确定函数具体的内容&#xff0c;即无法直接解出函数 Bernstein-Vazirani算法&#xff…...

jrebel IDEA 热部署

1 下载 2022.4.1 JRebel and XRebel - IntelliJ IDEs Plugin | Marketplace 2 选择下载好的zip 离线安装IDEA 插件 重启IDEA 3 打开 [Preference -> JRebel & XRebel] 菜单&#xff0c;输入 GUID address 为 https://jrebel.qekang.com/1e67ec1b-122f-4708-87d…...

pdf拆分成各个小pdf的方法

背景:由于某些缘故,一个大的pdf需要拆分成页数少的pdf,或者pdf需要去掉指定页,那么就有必要对pdf进行重新编辑,这里需要用到一个库,直接进行操作即可。 当使用Python时,可以使用PyMuPDF库来拆分PDF文件。以下是一个示例代码, import fitz # PyMuPDF def split_pdf(i…...

IntelliJ IDEA 常用快捷键一览表(通用型,提高编写速度,类结构、查找和查看源码,替换与关闭,调整格式)

文章目录 IntelliJ IDEA 常用快捷键一览表1-IDEA的日常快捷键第1组&#xff1a;通用型第2组&#xff1a;提高编写速度&#xff08;上&#xff09;第3组&#xff1a;提高编写速度&#xff08;下&#xff09;第4组&#xff1a;类结构、查找和查看源码第5组&#xff1a;查找、替换…...

MSVS C# Matlab的混合编程系列2 - 构建一个复杂(含多个M文件)的动态库:

前言: 本节我们尝试将一个有很多函数和文件的Matlab算法文件集成到C#的项目里面。 本文缩语: MT = Matlab 问题提出: 1 我们有一个比较复杂的Matlab文件: 这个MATLAB的算法,写了很多的算法函数在其他的M文件里面,这样,前面博客的方法就不够用了。会报错: 解决办法如下…...

上位机图像处理和嵌入式模块部署(qt图像处理)

【 声明&#xff1a;版权所有&#xff0c;欢迎转载&#xff0c;请勿用于商业用途。 联系信箱&#xff1a;feixiaoxing 163.com】 很多人一想到图像处理&#xff0c;本能的第一反应就是opencv&#xff0c;这也没有错。但是呢&#xff0c;这里面还是有一个问题的&#xff0c;不知…...

AI教我学编程之C#类的实例化与访问修饰符

前言 在这篇文章中&#xff0c;我将带大家深入了解C#编程语言的核心概念&#xff0c;包括类的实例化、访问修饰符的应用&#xff0c;以及C#中不同数据类型的默认值。我会通过逐步分析和具体实例&#xff0c;详细解释如何在C#中正确创建和操作对象&#xff0c;并探讨如何通过访…...

【笔记】Blender4.0建模入门-3物体的基本操作

Blender入门 ——邵发 3.1 物体的移动 演示&#xff1a; 1、选中一个物体 2、选中移动工具 3、移动 - 沿坐标轴移动 - 在坐标平面内移动 - 自由移动&#xff08;不好控制&#xff09; 选中物体&#xff1a;右上的大纲窗口&#xff0c;点击物体名称&#xff0c;物体的轮…...

一文详解 Berachain 测试网:全面介绍与教程,bitget wallet教程

什么是Berachain&#xff1f; Berachain&#xff08;web3.bitget.com/zh-CN/assets/berachain-wallet&#xff09;是一种尖端区块链技术&#xff0c;使用 Cosmos SDK 构建的 Layer-1&#xff0c;兼容以太坊虚拟机&#xff08;EVM&#xff09;。它基于一种独特的概念&#xff0c…...

小程序使用echarts图表-雷达图

本文介绍下小程序中如何使用echarts 如果是通过npm安装&#xff0c;这样是全部安装的&#xff0c;体积有点大 我这边是使用echarts中的一个组件来实现的&#xff0c;下边是具体流程&#xff0c;实际效果是没有外边的红色边框的&#xff0c;加红色边框的效果是这篇说明 1.echa…...

MacM1Pro Parallels19.1.0 CentOS7.9 Install PostgrepSQL

相关阅读 MacM1Pro安装 Parallels Desktop 19.1.0 https://blog.csdn.net/qq_41594280/article/details/135420241 MacM1Pro Parallels安装Parallels Tools https://blog.csdn.net/qq_41594280/article/details/135398780 MacM1Pro Parallels安装CentOS7.9 https://blog.csdn.n…...

Golang 中如何实现 Set

在Go编程中&#xff0c;数据结构的选择对解决问题至关重要。本文将探讨如何在 GO 中实现 set 和 bitset 两种数据结构&#xff0c;以及它们在Go中的应用场景。 Go 的数据结构 Go 内置的数据结构并不多。工作中&#xff0c;我们最常用的两种数据结构分别是 slice 和 map&#…...

记录一下uniapp 集成腾讯im特别卡(已解决)

uniapp的项目运行在微信小程序 , 安卓 , ios手机三端 , 之前这个项目集成过im,不过版本太老了,0.x的版本, 现在需要添加客服功能,所以就升级了 由于是二开 , 也为了方便 , 沿用之前的webview嵌套腾讯IM的方案 , 选用uniapp集成ui ,升级之后所有安卓用户反馈点击进去特别卡,几…...

React16源码: React中的updateHostRoot的源码实现

HostRoot 的更新 1 &#xff09;概述 HostRoot 是一个比较特殊的节点, 因为在一个react应用当中它只会有一个 HostRoot, 它对应的 Fiber 对象是我们的 RootFiber 对象重点在于它的更新过程 2 &#xff09;源码 定位到 packages/react-reconciler/src/ReactFiberBeginWork.js…...

Template -- React

React 版本 Node 21.6.0Npm 10.2.4 项目 创建 npm init vite 项目名称reactjsnpm inpm run dev 依赖 npm i axios # 网络 npm i antd --save # UI npm i ant-design/icons npm i react-router-dom # 路由npm i sass -D # …...

黑马Mybatis

Mybatis 表现层&#xff1a;页面展示 业务层&#xff1a;逻辑处理 持久层&#xff1a;持久数据化保存 在这里插入图片描述 Mybatis快速入门 ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/6501c2109c4442118ceb6014725e48e4.png //logback.xml <?xml ver…...

8k长序列建模,蛋白质语言模型Prot42仅利用目标蛋白序列即可生成高亲和力结合剂

蛋白质结合剂&#xff08;如抗体、抑制肽&#xff09;在疾病诊断、成像分析及靶向药物递送等关键场景中发挥着不可替代的作用。传统上&#xff0c;高特异性蛋白质结合剂的开发高度依赖噬菌体展示、定向进化等实验技术&#xff0c;但这类方法普遍面临资源消耗巨大、研发周期冗长…...

【位运算】消失的两个数字(hard)

消失的两个数字&#xff08;hard&#xff09; 题⽬描述&#xff1a;解法&#xff08;位运算&#xff09;&#xff1a;Java 算法代码&#xff1a;更简便代码 题⽬链接&#xff1a;⾯试题 17.19. 消失的两个数字 题⽬描述&#xff1a; 给定⼀个数组&#xff0c;包含从 1 到 N 所有…...

定时器任务——若依源码分析

分析util包下面的工具类schedule utils&#xff1a; ScheduleUtils 是若依中用于与 Quartz 框架交互的工具类&#xff0c;封装了定时任务的 创建、更新、暂停、删除等核心逻辑。 createScheduleJob createScheduleJob 用于将任务注册到 Quartz&#xff0c;先构建任务的 JobD…...

Frozen-Flask :将 Flask 应用“冻结”为静态文件

Frozen-Flask 是一个用于将 Flask 应用“冻结”为静态文件的 Python 扩展。它的核心用途是&#xff1a;将一个 Flask Web 应用生成成纯静态 HTML 文件&#xff0c;从而可以部署到静态网站托管服务上&#xff0c;如 GitHub Pages、Netlify 或任何支持静态文件的网站服务器。 &am…...

镜像里切换为普通用户

如果你登录远程虚拟机默认就是 root 用户&#xff0c;但你不希望用 root 权限运行 ns-3&#xff08;这是对的&#xff0c;ns3 工具会拒绝 root&#xff09;&#xff0c;你可以按以下方法创建一个 非 root 用户账号 并切换到它运行 ns-3。 一次性解决方案&#xff1a;创建非 roo…...

ETLCloud可能遇到的问题有哪些?常见坑位解析

数据集成平台ETLCloud&#xff0c;主要用于支持数据的抽取&#xff08;Extract&#xff09;、转换&#xff08;Transform&#xff09;和加载&#xff08;Load&#xff09;过程。提供了一个简洁直观的界面&#xff0c;以便用户可以在不同的数据源之间轻松地进行数据迁移和转换。…...

NFT模式:数字资产确权与链游经济系统构建

NFT模式&#xff1a;数字资产确权与链游经济系统构建 ——从技术架构到可持续生态的范式革命 一、确权技术革新&#xff1a;构建可信数字资产基石 1. 区块链底层架构的进化 跨链互操作协议&#xff1a;基于LayerZero协议实现以太坊、Solana等公链资产互通&#xff0c;通过零知…...

k8s业务程序联调工具-KtConnect

概述 原理 工具作用是建立了一个从本地到集群的单向VPN&#xff0c;根据VPN原理&#xff0c;打通两个内网必然需要借助一个公共中继节点&#xff0c;ktconnect工具巧妙的利用k8s原生的portforward能力&#xff0c;简化了建立连接的过程&#xff0c;apiserver间接起到了中继节…...

MySQL账号权限管理指南:安全创建账户与精细授权技巧

在MySQL数据库管理中&#xff0c;合理创建用户账号并分配精确权限是保障数据安全的核心环节。直接使用root账号进行所有操作不仅危险且难以审计操作行为。今天我们来全面解析MySQL账号创建与权限分配的专业方法。 一、为何需要创建独立账号&#xff1f; 最小权限原则&#xf…...