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

中间件-------RabbitMQ

同步和异步

异步调用

MQ

MQ优势:①服务解耦   ②异步调用   ③流量削峰

结构

消息模型

 RabbitMQ入门案例,实现消息发送和消息接收

生产者:
public class PublisherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码factory.setHost("192.168.136.132");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("itcast");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "hello, rabbitmq!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功:【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}
消费者:
public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码factory.setHost("192.168.136.132");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("itcast");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message = new String(body);System.out.println("接收到消息:【" + message + "】");}});System.out.println("等待接收消息。。。。");}
}

SpringAMQP

引入依赖

        <!--AMQP依赖,包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

普通队列 

 第一步:publisher服务配置文件,发消息

spring:rabbitmq:host: 192.168.136.132port: 5672username: itcastpassword: 123321virtual-host: /
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {//获取RabbitTemplateAPI@Resourceprivate RabbitTemplate rabbitTemplate;@Testpublic void test(){String queueName = "simple.queue";String message = "hello SpringAMQP";//使用API传入队列名和消息即可直接发送rabbitTemplate.convertAndSend(queueName,message);}}

第二步:Consumer服务配置信息监听消息

spring:rabbitmq:host: 192.168.136.132port: 5672username: itcastpassword: 123321virtual-host: /
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;//定义一个监听类去监听消息
@Component
public class SpringRabbitListener {@RabbitListener(queues = "simple.queue")public void ListenSimpleQueue(String msg){System.out.println("msg = " + msg);}
}


Work Queue队列

多个消费者绑定到同一个队列,可以通过prefetch来控制消费者消息预取的数量

第一步: 生产者发送消息

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {//获取RabbitTemplateAPI@Resourceprivate RabbitTemplate rabbitTemplate;@Testpublic void test01() throws InterruptedException {String queueName = "simple.queue";String message = "hello SpringAMQP--";for (int i = 0; i < 50; i++) {//使用API传入队列名和消息即可直接发送rabbitTemplate.convertAndSend(queueName,message+i);Thread.sleep(20);}}}

第二步:消费者设置多个监听消息

@Component
public class SpringRabbitListener {@RabbitListener(queues = "simple.queue")public void ListenWorkQueue(String msg) throws InterruptedException {System.out.println("消费者一接收到消息---- = " + msg + LocalDateTime.now());Thread.sleep(20);}@RabbitListener(queues = "simple.queue")public void ListenWorkQueue01(String msg) throws InterruptedException {System.out.println("消费者二接收到消息---- = " + msg + LocalDateTime.now());Thread.sleep(200);}
}

 第三步:消费者可通过prehtch设置消息预取数量

spring:rabbitmq:host: 192.168.136.132port: 5672username: itcastpassword: 123321virtual-host: /listener:simple:prefetch: 1


发布-订阅模型

Fanout广播交换机 --->多个队列收到交换机的消息

第一步:Consumer声明交换机,队列并进行绑定。
@Configuration
public class FanoutConfig {//声明交换机@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("itcast.fanout");}//声明队列1@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}//绑定队列1到交换机上@Beanpublic Binding fanoutBanding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//声明队列2@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}//绑定队列2到交换机上@Beanpublic Binding fanoutBanding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}
 第二步:Consumer进行监听消息
@Component
public class SpringRabbitListener {@RabbitListener(queues = "fanout.queue1")public void ListenSimpleQueue1(String msg){System.out.println("消费者接收到fanout.queue1的消息 = " + msg);}@RabbitListener(queues = "fanout.queue2")public void ListenSimpleQueue2(String msg){System.out.println("消费者接收到fanout.queue2的消息 = " + msg);}
}
第三步:Publisher向交换机发送消息
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {//获取RabbitTemplateAPI@Resourceprivate RabbitTemplate rabbitTemplate;@Testpublic void testExchange() {//声明交换机名称String exchangeName = "itcast.fanout";//消息String message = "Hello Everyone";//发送消息rabbitTemplate.convertAndSend(exchangeName,"",message);}
}

Direct路由交换机 --->将消息发给指定key的队列

第一步:在Listener中声明队列,交换机以及key
@Component
public class SpringRabbitListener {//声明队列1,交换机以及队列1的bindingKey@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "derict.queue1"),exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),key = {"red","blue"}))public void ListenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息 = " + msg);}//声明队列2,交换机以及队列2的bindingKey@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "derict.queue2"),exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),key = {"red","yellow"}))public void ListenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue2的消息 = " + msg);}
}
第二步:向指定key的队列发送消息
    @Testpublic void testDirect() {//声明交换机名称String exchangeName = "itcast.direct";//消息String message = "Hello Blue!!";//发送消息,指定交换机,队列以及要发送的keyrabbitTemplate.convertAndSend(exchangeName,"red",message);}

Topic主题交换机 ---->key必须是多个单词列表,统一主题,支持通配符

 第一步:在Listener中声明队列,交换机以及通配符key
@Component
public class SpringRabbitListener {//声明队列2的交换机,队列以及通配符key@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),key = "china.#"))public void ListenTopicQueue1(String msg){System.out.println("消费者接收到topic.queue1的消息 = " + msg);}//声明队列2的交换机,队列以及通配符key@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),key = "#.news"))public void ListenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息 = " + msg);}}
第二步:向主题通配符发送消息

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {//获取RabbitTemplateAPI@Resourceprivate RabbitTemplate rabbitTemplate;@Testpublic void testTopic() {//声明交换机名称String exchangeName = "itcast.topic";//消息String message = "Hello China!!";//发送消息rabbitTemplate.convertAndSend(exchangeName,"china.news",message);}
}

消息转换器

 RabbitMQ发的消息体都是Object类型,所有还可以发送对象数据。而且默认的消息转换器是MessageConverter实现的,当使用的是Map数据类型时,就会序列化成很多字节,所以推荐使用JSON的序列化和反序列化,直接修改默认的MessageConverter的类型

        <dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency>
@Beanpublic MessageConverter messageConverter(){return  new Jackson2JsonMessageConverter();}

对于RabbitMQ高级部分:死信队列,延迟队列,发布确认,幂等性,优先,惰性队列等有时间再学

相关文章:

中间件-------RabbitMQ

同步和异步 异步调用 MQ MQ优势&#xff1a;①服务解耦 ②异步调用 ③流量削峰 结构 消息模型 RabbitMQ入门案例&#xff0c;实现消息发送和消息接收 生产者&#xff1a; public class PublisherTest {Testpublic void testSendMessage() throws IOException, TimeoutExce…...

flink Data Source数据源

flink Data Source数据源 Source 并行度 非并行&#xff1a;并行度只能为1 并行 基于集合的Source fromElements package com.pxj.sx.flink; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.RestOptions; import org.ap…...

网络七层模型与云计算中的网络服务

网络七层模型&#xff0c;也称为OSI&#xff08;Open System Interconnection&#xff09;模型&#xff0c;是由国际标准化组织&#xff08;ISO&#xff09;制定的一个概念性框架&#xff0c;用于描述网络通信过程中信息是如何被封装、传输和解封装的。这一模型将复杂的网络通信…...

word一按空格就换行怎么办?word文本之间添加空格就换行怎么办?

如上图&#xff0c;无法在Connection和con之间添加空格&#xff0c;一按空格就会自动换行。 第一步&#xff1a;选中文本&#xff0c;打开段落。 第二步&#xff1a;点击中文版式&#xff0c;勾选允许西文在单词中间换行。 确定之后就解决一按空格就自动换行啦&#xff01;...

Python 遍历字典的方法,你都掌握了吗

Python中的字典是一种非常灵活的数据结构&#xff0c;它允许通过键来存储和访问值。在处理字典时&#xff0c;经常需要遍历字典中的元素&#xff0c;以下是几种常见的遍历字典的方法。 1. 使用 for 循环直接遍历字典的键 字典的键是唯一的&#xff0c;可以直接通过 for 循环来…...

MySQL 8.4.0 LTS 变更解析:I_S 表、权限、关键字和客户端

↑ 关注“少安事务所”公众号&#xff0c;欢迎⭐收藏&#xff0c;不错过精彩内容~ MySQL 8.4.0 LTS 已经发布 &#xff0c;作为发版模型变更后的第一个长期支持版本&#xff0c;注定要承担未来生产环境的重任&#xff0c;那么这个版本都有哪些新特性、变更&#xff0c;接下来少…...

LeetCode 124 —— 二叉树中的最大路径和

阅读目录 1. 题目2. 解题思路3. 代码实现 1. 题目 2. 解题思路 二叉树的问题首先我们要想想是否能用递归来解决&#xff0c;本题也不例外&#xff0c;而递归的关键是找到子问题。 我们首先来看看一棵最简单的树&#xff0c;也就是示例 1。这样的一棵树总共有六条路径&#xf…...

美甲店会员预约系统管理小程序的作用是什么

女性爱美体现在方方面面&#xff0c;美丽好看的指甲也不能少&#xff0c;市场中美甲店、小摊不少&#xff0c;也跑出了不少连锁品牌&#xff0c;70后到00后&#xff0c;每个层级都有不少潜在客户&#xff0c;商家需要获取和完善转化路径&#xff0c;不断提高品牌影响力与自身内…...

..堆..

堆 堆是完全二叉树&#xff0c;即除了最后一列之外&#xff0c;上面的每一层都是满的&#xff08;左右严格对称且每个节点都满子节点&#xff09; 最后一列从左向右排序。 默认大根堆&#xff1a;每一个节点都大于其左右儿子&#xff0c;根节点就是整个数据结构的最大值 pr…...

【LLM多模态】综述Visual Instruction Tuning towards General-Purpose Multimodal Model

note 文章目录 note论文1. 论文试图解决什么问题2. 这是否是一个新的问题3. 这篇文章要验证一个什么科学假设4. 有哪些相关研究&#xff1f;如何归类&#xff1f;谁是这一课题在领域内值得关注的研究员&#xff1f;5. 论文中提到的解决方案之关键是什么&#xff1f;6. 论文中的…...

探索Linux中的神奇工具:重定向符的妙用

探索Linux中的神奇工具&#xff1a;重定向符的妙用 在Linux系统中&#xff0c;重定向符是一个强大的工具&#xff0c;用于控制命令的输入和输出&#xff0c;实现数据流的定向。本文将详细介绍重定向符的基本用法和一些实用技巧&#xff0c;帮助读者更好地理解和运用这个功能。…...

Kubernetes 文档 / 概念 / 工作负载 / 工作负载管理 / Job

Kubernetes 文档 / 概念 / 工作负载 / 工作负载管理 / Job 此文档从 Kubernetes 官网摘录 中文地址 英文地址 Job 会创建一个或者多个 Pod&#xff0c;并将继续重试 Pod 的执行&#xff0c;直到指定数量的 Pod 成功终止。 随着 Pod 成功结束&#xff0c;Job 跟踪记录成功完成的…...

办公自动化-Python如何提取Word标题并保存到Excel中?

办公自动化-Python如何提取Word标题并保存到Excel中&#xff1f; 应用场景需求分析实现思路实现过程安装依赖库打开需求文件获取word中所有标题去除不需要的标题创建工作簿和工作表分割标题功能名称存入测试对象GN-TC需求标识符存入测试项标识存入需求标识符 完整源码实现效果学…...

基于Java、SpringBoot和uniapp在线考试系统安卓APP和微信小程序

摘要 基于Java、SpringBoot和uniapp的在线考试系统安卓APP微信小程序是一种结合了现代Web开发技术和移动应用技术的解决方案&#xff0c;旨在为教育机构提供一个方便、高效和灵活的在线考试平台。该系统采用Java语言进行后端开发&#xff0c;使用SpringBoot框架简化企业级应用…...

抖音a-bogus加密解析(三)

要补的环境我给提示&#xff0c;大家自行操作&#xff0c;出了问题就是因为缺环境&#xff0c;没补好 window global; // reading _u未定义 window.requestAnimationFrame function () {} // XMLHttpRequest 未定义 window.XMLHttpRequest function () {} window.onwheelx …...

IS-IS DIS

原理概述 OSPF 协议支持4种网络类型&#xff0c; IS-IS 协议只支持两种网络类型&#xff0c;即广播网络和点到点网络。与 OSPF 协议相同&#xff0c; IS-IS 协议在广播网络中会将网络视为一个伪节点( Pseudonode &#xff0c;简称 PSN )&#xff0c;并选举出一台 DIS ( Designa…...

random和range

含义&#xff1a; random(1&#xff0c;10) 不包含10&#xff0c;用于生成随机数。它可以生成浮点数或整数&#xff0c;取决于具体的使用方式。 range(0&#xff0c;1) 不包含1&#xff0c;用于生成一个整数序列。它可以生成一个指定范围内的连续整数序列。 区别在于&#x…...

研二学妹面试字节,竟倒在了ThreadLocal上,这是不要应届生还是不要女生啊?

一、写在开头 今天和一个之前研二的学妹聊天&#xff0c;聊及她上周面试字节的情况&#xff0c;着实感受到了Java后端现在找工作的压力啊&#xff0c;记得在18&#xff0c;19年的时候&#xff0c;研究生计算机专业的学生&#xff0c;背背八股文找个Java开发工作毫无问题&#x…...

Golang:gammazero/deque是一个快速环形缓冲区deque(双端队列)实现

gammazero/deque是一个快速环形缓冲区deque&#xff08;双端队列&#xff09;实现。 文档 https://github.com/gammazero/deque 安装 go get github.com/gammazero/deque代码示例 先入先出队列 package mainimport ("fmt""github.com/gammazero/deque&quo…...

C++ 时间处理-统计函数运行时间

1. 关键词2. 问题3. 解决思路4. 代码实现 4.1. timecount.h4.2. timecount.cpp 5. 测试代码6. 运行结果7. 源码地址 1. 关键词 C 时间处理 统计函数运行时间 跨平台 2. 问题 C如何简单便捷地实现“函数运行时间的统计”功能&#xff1f; 3. 解决思路 类的构造函数&#x…...

SkyWalking 10.2.0 SWCK 配置过程

SkyWalking 10.2.0 & SWCK 配置过程 skywalking oap-server & ui 使用Docker安装在K8S集群以外&#xff0c;K8S集群中的微服务使用initContainer按命名空间将skywalking-java-agent注入到业务容器中。 SWCK有整套的解决方案&#xff0c;全安装在K8S群集中。 具体可参…...

Oracle查询表空间大小

1 查询数据库中所有的表空间以及表空间所占空间的大小 SELECTtablespace_name,sum( bytes ) / 1024 / 1024 FROMdba_data_files GROUP BYtablespace_name; 2 Oracle查询表空间大小及每个表所占空间的大小 SELECTtablespace_name,file_id,file_name,round( bytes / ( 1024 …...

376. Wiggle Subsequence

376. Wiggle Subsequence 代码 class Solution { public:int wiggleMaxLength(vector<int>& nums) {int n nums.size();int res 1;int prediff 0;int curdiff 0;for(int i 0;i < n-1;i){curdiff nums[i1] - nums[i];if( (prediff > 0 && curdif…...

Nuxt.js 中的路由配置详解

Nuxt.js 通过其内置的路由系统简化了应用的路由配置&#xff0c;使得开发者可以轻松地管理页面导航和 URL 结构。路由配置主要涉及页面组件的组织、动态路由的设置以及路由元信息的配置。 自动路由生成 Nuxt.js 会根据 pages 目录下的文件结构自动生成路由配置。每个文件都会对…...

Psychopy音频的使用

Psychopy音频的使用 本文主要解决以下问题&#xff1a; 指定音频引擎与设备&#xff1b;播放音频文件 本文所使用的环境&#xff1a; Python3.10 numpy2.2.6 psychopy2025.1.1 psychtoolbox3.0.19.14 一、音频配置 Psychopy文档链接为Sound - for audio playback — Psy…...

解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错

出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上&#xff0c;所以报错&#xff0c;到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本&#xff0c;cu、torch、cp 的版本一定要对…...

成都鼎讯硬核科技!雷达目标与干扰模拟器,以卓越性能制胜电磁频谱战

在现代战争中&#xff0c;电磁频谱已成为继陆、海、空、天之后的 “第五维战场”&#xff0c;雷达作为电磁频谱领域的关键装备&#xff0c;其干扰与抗干扰能力的较量&#xff0c;直接影响着战争的胜负走向。由成都鼎讯科技匠心打造的雷达目标与干扰模拟器&#xff0c;凭借数字射…...

NPOI操作EXCEL文件 ——CAD C# 二次开发

缺点:dll.版本容易加载错误。CAD加载插件时&#xff0c;没有加载所有类库。插件运行过程中用到某个类库&#xff0c;会从CAD的安装目录找&#xff0c;找不到就报错了。 【方案2】让CAD在加载过程中把类库加载到内存 【方案3】是发现缺少了哪个库&#xff0c;就用插件程序加载进…...

【前端异常】JavaScript错误处理:分析 Uncaught (in promise) error

在前端开发中&#xff0c;JavaScript 异常是不可避免的。随着现代前端应用越来越多地使用异步操作&#xff08;如 Promise、async/await 等&#xff09;&#xff0c;开发者常常会遇到 Uncaught (in promise) error 错误。这个错误是由于未正确处理 Promise 的拒绝&#xff08;r…...

Python 训练营打卡 Day 47

注意力热力图可视化 在day 46代码的基础上&#xff0c;对比不同卷积层热力图可视化的结果 import torch import torch.nn as nn import torch.optim as optim from torchvision import datasets, transforms from torch.utils.data import DataLoader import matplotlib.pypl…...