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

RabbitMQ的部分模式

1发布订阅模式

发送者

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class PublishProduct {public static void main(String[] args) {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的地址,我用的redis,RabbitMQ,现在redis里安装RabbitMQfactory.setHost("192.168.74.75");Connection connection = null;Channel channel = null;try {connection = factory.newConnection();// 创建一个通道channel = connection.createChannel();//创建交换机channel.exchangeDeclare("qy172-fanout-exchange", BuiltinExchangeType.FANOUT, true);//创建队列,如果存在则不会创建channel.queueDeclare("qy172-publish-queue01", true, false, false, null);channel.queueDeclare("qy172-publish-queue02", true, false, false, null);//交互机和队列绑定channel.queueBind("qy172-publish-queue01", "qy172-fanout-exchange", "");channel.queueBind("qy172-publish-queue02", "qy172-fanout-exchange", "");// 创建消息内容HashMap<String, Object> map = new HashMap<>();map.put("name", "张三");map.put("age", "22");
//把数据给交换机,让他分发给队列channel.basicPublish("qy172-fanout-exchange", "", null, JSON.toJSONBytes(map));System.out.println("发送成功");} catch (IOException e) {// 发生 IO 异常时抛出运行时异常throw new RuntimeException(e);} catch (TimeoutException e) {// 发生超时异常时抛出运行时异常throw new RuntimeException(e);} finally {if (channel != null) {try {// 关闭通道channel.close();} catch (IOException | TimeoutException e) {// 发生 IO 或超时异常时抛出运行时异常throw new RuntimeException(e);}}if (connection != null) {try {// 关闭连接connection.close();} catch (IOException e) {// 发生 IO 异常时抛出运行时异常throw new RuntimeException(e);}}}}
}

2订阅个订阅者

订阅者1

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
public class Consumer01 {public static void main(String[] args) throws Exception {// 创建连接工厂对象ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的主机地址为 "192.168.74.75"factory.setHost("192.168.74.75");Connection connection = factory.newConnection();// 创建一个 RabbitMQ 连接Channel channel = connection.createChannel();// 创建一个通道,用于与 RabbitMQ 之间的通信com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {// 创建一个消费者对象,并重写其方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 消费消息的处理方法String json = new String(body);// 将消息内容转换为字符串Map map = JSON.parseObject(json, Map.class);// 使用 JSON 解析成 Map 对象System.out.println("消息内容Consumer01"+map);// 输出消息内容}};channel.basicConsume("qy172-publish-queue01",true,consumer);}
}

订阅者2

package com.aaa;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
public class Consumer02 {public static void main(String[] args) {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.74.75");try {Connection connection = factory.newConnection();Channel channel = connection.createChannel();com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String json = new String(body);Map map = JSON.parseObject(json, Map.class);System.out.println("消息内容Consumer02" + map);}};//订阅者2channel.basicConsume("qy172-publish-queue02",true,consumer);} catch (IOException | TimeoutException e) {// 处理连接、通道创建或消费消息时可能抛出的异常e.printStackTrace();}}
}

2路由模式

发送者

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class PublishProduct {public static void main(String[] args) {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的地址,我用的redis,RabbitMQ,现在redis里安装RabbitMQfactory.setHost("192.168.74.75");Connection connection = null;Channel channel = null;try {connection = factory.newConnection();// 创建一个通道channel = connection.createChannel();//创建交换机,channel.exchangeDeclare("qy172-router-exchange", BuiltinExchangeType.DIRECT, true);//创建队列,如果存在则不会创建channel.queueDeclare("qy172-router-queue01", true, false, false, null);channel.queueDeclare("qy172-router-queue02", true, false, false, null);//交互机和队列绑定channel.queueBind("qy172-router-queue01", "qy172-router-exchange", "error");channel.queueBind("qy172-router-queue02", "qy172-router-exchange", "error");channel.queueBind("qy172-router-queue02", "qy172-router-exchange", "info");channel.queueBind("qy172-router-queue02", "qy172-router-exchange", "warning");// 创建消息内容HashMap<String, Object> map = new HashMap<>();map.put("name", "张三");map.put("age", "22");//把数据给交换机,让他分发给队列channel.basicPublish("qy172-router-exchange","error",null,JSON.toJSONBytes(map));
//            channel.basicPublish("qy172-router-exchange","info",null,JSON.toJSONBytes(map));System.out.println("发送成功");} catch (IOException e) {// 发生 IO 异常时抛出运行时异常throw new RuntimeException(e);} catch (TimeoutException e) {// 发生超时异常时抛出运行时异常throw new RuntimeException(e);} finally {if (channel != null) {try {// 关闭通道channel.close();} catch (IOException | TimeoutException e) {// 发生 IO 或超时异常时抛出运行时异常throw new RuntimeException(e);}}if (connection != null) {try {// 关闭连接connection.close();} catch (IOException e) {// 发生 IO 异常时抛出运行时异常throw new RuntimeException(e);}}}}
}

接收者1

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
public class Consumer01 {public static void main(String[] args) throws Exception {// 创建连接工厂对象ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的主机地址为 "192.168.74.75"factory.setHost("192.168.74.75");Connection connection = factory.newConnection();// 创建一个 RabbitMQ 连接Channel channel = connection.createChannel();// 创建一个通道,用于与 RabbitMQ 之间的通信com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {// 创建一个消费者对象,并重写其方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 消费消息的处理方法String json = new String(body);// 将消息内容转换为字符串Map map = JSON.parseObject(json, Map.class);// 使用 JSON 解析成 Map 对象System.out.println("消息内容Consumer01"+map);// 输出消息内容}};channel.basicConsume("qy172-router-queue01",true,consumer);}
}

接收者2

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
public class Consumer01 {public static void main(String[] args) throws Exception {// 创建连接工厂对象ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的主机地址为 "192.168.74.75"factory.setHost("192.168.74.75");Connection connection = factory.newConnection();// 创建一个 RabbitMQ 连接Channel channel = connection.createChannel();// 创建一个通道,用于与 RabbitMQ 之间的通信com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {// 创建一个消费者对象,并重写其方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 消费消息的处理方法String json = new String(body);// 将消息内容转换为字符串Map map = JSON.parseObject(json, Map.class);// 使用 JSON 解析成 Map 对象System.out.println("消息内容Consumer01"+map);// 输出消息内容}};channel.basicConsume("qy172-router-queue01",true,consumer);}
}

3主题模式

发送者

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class PublishProduct {public static void main(String[] args) {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的地址,我用的redis,RabbitMQ,现在redis里安装RabbitMQfactory.setHost("192.168.74.75");Connection connection = null;Channel channel = null;try {connection = factory.newConnection();// 创建一个通道channel = connection.createChannel();//创建交换机,channel.exchangeDeclare("qy172-topic-exchange", BuiltinExchangeType.TOPIC, true);//创建队列,如果存在则不会创建channel.queueDeclare("qy172-topic-queue01", true, false, false, null);channel.queueDeclare("qy172-topic-queue02", true, false, false, null);//交互机和队列绑定//主题匹配给这个channel.queueBind("qy172-topic-queue01", "qy172-topic-exchange", "*.orange.*");//主题,也匹配给这个channel.queueBind("qy172-topic-queue02", "qy172-topic-exchange", "*.*.rabbit");channel.queueBind("qy172-topic-queue02", "qy172-topic-exchange", "lazy.#");// 创建消息内容HashMap<String, Object> map = new HashMap<>();map.put("name", "张三");map.put("age", "22");//把数据给交换机,让他分发给队列channel.basicPublish("qy172-topic-exchange","lazy.orange.rabbit",null,JSON.toJSONBytes(map));System.out.println("发送成功");} catch (IOException e) {// 发生 IO 异常时抛出运行时异常throw new RuntimeException(e);} catch (TimeoutException e) {// 发生超时异常时抛出运行时异常throw new RuntimeException(e);} finally {if (channel != null) {try {// 关闭通道channel.close();} catch (IOException | TimeoutException e) {// 发生 IO 或超时异常时抛出运行时异常throw new RuntimeException(e);}}if (connection != null) {try {// 关闭连接connection.close();} catch (IOException e) {// 发生 IO 异常时抛出运行时异常throw new RuntimeException(e);}}}}
}

接收者1

package org.example;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
public class Consumer01 {public static void main(String[] args) throws Exception {// 创建连接工厂对象ConnectionFactory factory = new ConnectionFactory();// 设置 RabbitMQ 服务器的主机地址为 "192.168.74.75"factory.setHost("192.168.74.75");Connection connection = factory.newConnection();// 创建一个 RabbitMQ 连接Channel channel = connection.createChannel();// 创建一个通道,用于与 RabbitMQ 之间的通信com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {// 创建一个消费者对象,并重写其方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 消费消息的处理方法String json = new String(body);// 将消息内容转换为字符串Map map = JSON.parseObject(json, Map.class);// 使用 JSON 解析成 Map 对象System.out.println("消息内容Consumer01"+map);// 输出消息内容}};channel.basicConsume("qy172-topic-queue01",true,consumer);}
}

接收者2

package com.aaa;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
public class Consumer02 {public static void main(String[] args) {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.74.75");try {Connection connection = factory.newConnection();Channel channel = connection.createChannel();com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String json = new String(body);Map map = JSON.parseObject(json, Map.class);System.out.println("消息内容Consumer02" + map);}};//订阅者2channel.basicConsume("qy172-topic-queue02",true,consumer);} catch (IOException | TimeoutException e) {// 处理连接、通道创建或消费消息时可能抛出的异常e.printStackTrace();}}
}

相关文章:

RabbitMQ的部分模式

1发布订阅模式 发送者 package org.example; import com.alibaba.fastjson.JSON; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import ja…...

提取单选框的值,并通过ajax传值到后台

<!DOCTYPE html> <html lang"zh" xmlns:th"http://www.thymeleaf.org" xmlns:shiro"http://www.pollix.at/thymeleaf/shiro"> <head><th:block th:include"include :: header(日库存更新提示)" /> </head&…...

Django创建多app应用

目录 1. 引言 2. 多app创建的两种方式 2.1 多个app结构 2.2 单个apps多个app 3. 最后 1. 引言 在平常业务开发中&#xff0c;我们遇到的功能可能会有很多&#xff0c;单个app的应用可能无法满足我们 这个时候&#xff0c;我们就需要多app应用&#xff0c;例如&#xff1a…...

如何反反爬虫

我们来讲最常见的反反爬虫方法 import requests r requests.get(网页网址) print(r.requests.headers) 一.使用简单的方法把请求头改为真的浏览器模式 import requests link网页地址 heraders{User-Agent:} rrequests.get(link,headersheaders) print(r.requsts.headers)我们…...

wireshark抓包之DNS协议

DNS协议 DNS协议的主要作用是将域名解析为对应的IP地址。当我们在浏览器中输入一个网址时&#xff0c;计算机需要通过DNS协议来查找该网址对应的IP地址&#xff0c;以便能够建立连接并访问目标资源。 DNS协议的工作流程大致如下&#xff1a; 用户的计算机或设备&#xff08;充…...

升级到 Java 21 是值得的

升级到 Java 21 是值得的 又到了一年中的这个时候——New Relic 的年度“State of the Java Ecosystem”调查结果出来了&#xff0c;我一如既往地深入研究了它。虽然我认为该报告做得很好并且提出了很好的问题&#xff0c;但我对有多少 Java 开发人员正在使用低版本感到沮丧。…...

C# 多线程

文章目录 C# 多线程进程与线程无参数的子线程带参数的子线程运行结果 销毁线程 Abort()运行结果 ThreadPool和Task运行结果 异步与同步运行结果 lock单线程运行结果 多线程运行结果 使用lock运行结果 C# 多线程 进程与线程 进程&#xff1a;进程就是一个应用程序&#xff0c;…...

快速安装sudachipy日语包

1、前往 https://rustup.rs 下载并安装 Rustup Linux系统可直接运行以下命令 Window系统需要去网站下载exe包 curl --proto https --tlsv1.2 -sSf https://sh.rustup.rs | sh2、安装 Rust 编译器 rustup install stable3、设置默认版本 rustup default stable4、重新安装 …...

蓝桥杯刷题day13——乘飞机【算法赛】

一、问题描述 等待登机的你看着眼前有老有小长长的队伍十分无聊&#xff0c;你突然想要知道&#xff0c;是否存在两个年龄相仿的乘客。每个乘客的年龄用一个 0 到 36500 的整数表示&#xff0c;两个乘客的年龄相差 365 以内就认为是相仿的。 具体来说&#xff0c;你有一个长度…...

大模型量化技术-BitsAndBytes

Transformers 量化技术 BitsAndBytes bitsandbytes是将模型量化为8位和4位的最简单选择。 8位量化将fp16中的异常值与int8中的非异常值相乘,将非异常值转换回fp16,然后将它们相加以返回fp16中的权重。这减少了异常值对模型性能产生的降级效果。4位量化进一步压缩了模型,并且…...

EasyExcel 复杂表头的导出(动态表头和静态表头)

问题&#xff1a;如图&#xff0c;1部分的表头是动态的根据日期变化&#xff0c;2部分是数据库对应的字段&#xff0c;静态不变的&#xff1b; 解决方案&#xff1a;如果不看1的部分&#xff0c;2部分内容可以根据实体类注解的方式导出&#xff0c;那么我们是不是可以先将动态表…...

centos7 fatal error: curl/curl.h: No such file or directory

若编译遇到此问题&#xff0c;可以查看环境是否libcurl库 yum list installed | grep libcurl 发现未安装libcurl库 执行libcurl库的安装命令&#xff1a; 1.对于Debian/Ubuntu系统&#xff1a; sudo apt-get install libcurl4-openssl-dev 2.对于RHEL/CentOS系统&#xf…...

【Linux】自定义协议+序列化+反序列化

自定义协议序列化反序列化 1.再谈 "协议"2.Cal TCP服务端2.Cal TCP客户端4.Json 喜欢的点赞&#xff0c;收藏&#xff0c;关注一下把&#xff01; 1.再谈 “协议” 协议是一种 “约定”。在前面我们说过父亲和儿子约定打电话的例子&#xff0c;不过这是感性的认识&a…...

常见故障排查和优化

一、MySQL单实例故障排查 故障现象 1 ERROR 2002 (HY000): Cant connect to local MySQL server through socket /data/mysql/mysql.sock (2) 问题分析&#xff1a;以上情况一般都是数据库未启动或者数据库端口被防火墙拦截导致。 解决方法&#xff1a;启动数据库或者防火墙…...

选择华为HCIE培训机构有哪些注意事项

选择软件培训机构注意四点事项1、口碑&#xff1a;学员和社会人士对该机构的评价怎样&#xff1f; 口碑对于一个机构是十分重要的&#xff0c;这也是考量一个机构好不好的重要标准&#xff0c;包括社会评价和学员的评价和感言。誉天作为华为首批授权培训中心&#xff0c;一直致…...

python怎么处理txt

导入文件处理模块 import os 检测路径是否存在&#xff0c;存在则返回True&#xff0c;不存在则返回False os.path.exists("demo.txt") 如果你要创建一个文件并要写入内容 #如果demo.txt文件存在则会覆盖&#xff0c;并且demo.txt文件里面的内容被清空&#xff0c;如…...

SAMRTFORMS 转换PDF 发送邮件

最终成果&#xff1a; *&---------------------------------------------------------------------**& Report ZLC_FIND_EXIT*&---------------------------------------------------------------------**&根据T-CODE / 程序名查询出口、BADI增强*&-------…...

探讨在大数据体系中API的通信机制与工作原理

** 引言 关联阅读博客文章&#xff1a;深入解析大数据体系中的ETL工作原理及常见组件 关联阅读博客文章&#xff1a;深入理解HDFS工作原理&#xff1a;大数据存储和容错性机制解析 ** 在当今数字化时代&#xff0c;数据已经成为企业发展和决策的核心。随着数据规模的不断增长…...

算法打卡day23

今日任务&#xff1a; 1&#xff09;39. 组合总和 2&#xff09;40.组合总和II 3&#xff09;131.分割回文串 39. 组合总和 题目链接&#xff1a;39. 组合总和 - 力扣&#xff08;LeetCode&#xff09; 给定一个无重复元素的数组 candidates 和一个目标数 target &#xff0c;…...

每天五分钟深度学习:神经网络和深度学习有什么样的关系?

本文重点 神经网络是一种模拟人脑神经元连接方式的计算模型&#xff0c;通过大量神经元之间的连接和权重调整&#xff0c;实现对输入数据的处理和分析。而深度学习则是神经网络的一种特殊形式&#xff0c;它通过构建深层次的神经网络结构&#xff0c;实现对复杂数据的深度学习…...

如何在看板中体现优先级变化

在看板中有效体现优先级变化的关键措施包括&#xff1a;采用颜色或标签标识优先级、设置任务排序规则、使用独立的优先级列或泳道、结合自动化规则同步优先级变化、建立定期的优先级审查流程。其中&#xff0c;设置任务排序规则尤其重要&#xff0c;因为它让看板视觉上直观地体…...

《基于Apache Flink的流处理》笔记

思维导图 1-3 章 4-7章 8-11 章 参考资料 源码&#xff1a; https://github.com/streaming-with-flink 博客 https://flink.apache.org/bloghttps://www.ververica.com/blog 聚会及会议 https://flink-forward.orghttps://www.meetup.com/topics/apache-flink https://n…...

CRMEB 框架中 PHP 上传扩展开发:涵盖本地上传及阿里云 OSS、腾讯云 COS、七牛云

目前已有本地上传、阿里云OSS上传、腾讯云COS上传、七牛云上传扩展 扩展入口文件 文件目录 crmeb\services\upload\Upload.php namespace crmeb\services\upload;use crmeb\basic\BaseManager; use think\facade\Config;/*** Class Upload* package crmeb\services\upload* …...

【网络安全】开源系统getshell漏洞挖掘

审计过程&#xff1a; 在入口文件admin/index.php中&#xff1a; 用户可以通过m,c,a等参数控制加载的文件和方法&#xff0c;在app/system/entrance.php中存在重点代码&#xff1a; 当M_TYPE system并且M_MODULE include时&#xff0c;会设置常量PATH_OWN_FILE为PATH_APP.M_T…...

通过MicroSip配置自己的freeswitch服务器进行调试记录

之前用docker安装的freeswitch的&#xff0c;启动是正常的&#xff0c; 但用下面的Microsip连接不上 主要原因有可能一下几个 1、通过下面命令可以看 [rootlocalhost default]# docker exec -it freeswitch fs_cli -x "sofia status profile internal"Name …...

算法打卡第18天

从中序与后序遍历序列构造二叉树 (力扣106题) 给定两个整数数组 inorder 和 postorder &#xff0c;其中 inorder 是二叉树的中序遍历&#xff0c; postorder 是同一棵树的后序遍历&#xff0c;请你构造并返回这颗 二叉树 。 示例 1: 输入&#xff1a;inorder [9,3,15,20,7…...

Visual Studio Code 扩展

Visual Studio Code 扩展 change-case 大小写转换EmmyLua for VSCode 调试插件Bookmarks 书签 change-case 大小写转换 https://marketplace.visualstudio.com/items?itemNamewmaurer.change-case 选中单词后&#xff0c;命令 changeCase.commands 可预览转换效果 EmmyLua…...

C++_哈希表

本篇文章是对C学习的哈希表部分的学习分享 相信一定会对你有所帮助~ 那咱们废话不多说&#xff0c;直接开始吧&#xff01; 一、基础概念 1. 哈希核心思想&#xff1a; 哈希函数的作用&#xff1a;通过此函数建立一个Key与存储位置之间的映射关系。理想目标&#xff1a;实现…...

ubuntu清理垃圾

windows和ubuntu 双系统&#xff0c;ubuntu 150GB&#xff0c;开发用&#xff0c;基本不装太多软件。但是磁盘基本用完。 1、查看home目录 sudo du -h -d 1 $HOME | grep -v K 上面的命令查看$HOME一级目录大小&#xff0c;发现 .cache 有26GB&#xff0c;.local 有几个GB&am…...

SDU棋界精灵——硬件程序ESP32实现opus编码

一、 ​​音频处理框架​ 该项目基于Espressif的音频处理框架构建,核心组件包括 ESP-ADF 和 ESP-SR,以下是完整的音频处理框架实现细节: 1.核心组件 (1) 音频前端处理 (AFE - Audio Front-End) ​​main/components/audio_pipeline/afe_processor.c​​功能​​: 声学回声…...