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();}}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/787083.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

如何使用【知乎引流软件工具】进行有效引流

如何使用【知乎引流软件工具】进行有效引流

知乎&#xff0c;这个内容平台&#xff0c;无疑是品牌和创作者们的乐园。 品牌能在这里找到目标关键词&#xff0c;而创作者们的内容则可以吸引流量&#xff0c;助力各种项目。 别被那些所谓的爆粉秘籍迷惑了&#xff0c;真正的成功往往来自于扎实的内容和策略&#xff0c;而非…
阅读更多...
云原生（七）、Kubernetes初学 + 裸机搭建k8s集群

云原生（七）、Kubernetes初学 + 裸机搭建k8s集群

Kubernetes简介 Kubernetes&#xff08;通常简称为K8s&#xff09;是一个开源的容器编排平台&#xff0c;最初由Google设计和开发&#xff0c;现在由Cloud Native Computing Foundation&#xff08;CNCF&#xff09;维护。它旨在简化容器化应用程序的部署、扩展和管理。 Kube…
阅读更多...
IPC 进程间通信

IPC 进程间通信

IPC InterProcess Communication The concept of IPC Each process has a differnt user addess space,and local variables 各自看不见,so 进程间通信 need kernel(内核), so a buffer is opened in the kernel,process 1 copies data from user space to this buffer,and …
阅读更多...
Linux：运营商在网络中扮演的角色

Linux：运营商在网络中扮演的角色

文章目录 ip目前的问题ip目前的几种解决方案私有ipVS公有ip运营商再谈ip划分运营商的角度看ip 本篇总结的是运营商在网络中扮演的角色 ip目前的问题 在目前看来&#xff0c;ip最大的问题是ip号不够用了&#xff0c;那这个问题如何解决呢&#xff1f; 在之前的内容中有子网掩…
阅读更多...
朗之万方程，机器学习与液体中的粒子运动

朗之万方程，机器学习与液体中的粒子运动

目录 一、说明二、朗之万方程的诞生2.1 牛顿力学2.2 流体中的随机运动 三、小质量物体布朗运动方程四、布朗运动的Python代码五、稳定性讨论5.1 波尔兹曼分布5.2 梯度下降算法 六、随机梯度下降&#xff08;SGD&#xff09;和小批量梯度下降七、机器学习与物理&#xff0c;作为…
阅读更多...
加密软件VMProtect教程：使用脚本-功能

加密软件VMProtect教程：使用脚本-功能

VMProtect是新一代软件保护实用程序。VMProtect支持德尔菲、Borland C Builder、Visual C/C、Visual Basic&#xff08;本机&#xff09;、Virtual Pascal和XCode编译器。 同时&#xff0c;VMProtect有一个内置的反汇编程序&#xff0c;可以与Windows和Mac OS X可执行文件一起…
阅读更多...
最大连续1的个数||

最大连续1的个数||

又用到滑动窗口&#xff0c;关键在于要维持窗口内0的数量在k内 #include<stdio.h> #include<string.h>int Max(int a,int b){return (a>b)?a:b; }int findMaxConsecutiveOnes(int* nums,int numsSize,int k){int left 0;int right 0;int flag 0;//记录包含…
阅读更多...
系统设计理念：满足设计要求

系统设计理念：满足设计要求

在今天的文章中，我们将层层剥离并深入探讨系统设计要求的本质。 让我们了解系统设计的支柱以及如何创建强大且有弹性的应用程序。 什么是好的设计？ 在我们深入讨论技术细节之前，让我们先讨论一下什么是好的设计。 当我们谈论系统架构中的良好设计时，我们关注几个关键原则：…
阅读更多...
初始C语言最后一章《编译、链接与预处理详解》

初始C语言最后一章《编译、链接与预处理详解》

前言 感谢老铁们的陪伴和支持&#xff0c;初始C语言专栏在本章内容也是要结束了&#xff0c;这创作一路下来也是很不容易&#xff0c;如果大家对 Java 后端开发感兴趣&#xff0c;欢迎各位老铁来我的Java专栏&#xff01;当然了&#xff0c;我也会更新几章C语言实现简单的数据结…
阅读更多...
C++中的面向对象到底是什么

C++中的面向对象到底是什么

C中的面向对象到底是什么 对象嘛&#xff0c;就和大家都有的对象一样&#xff0c;两只眼睛、一个嘴巴、两条腿…… 对不起跑题了&#xff0c;C的面向对象中的对象可不是显示中的对象哦&#xff0c;但是有一些相似之处&#xff0c;有对象的同学可以参考着去学习C面向对象的概念…
阅读更多...
HackTheBox-Machines--Builder

HackTheBox-Machines--Builder

文章目录 1 端口扫描2 测试思路3 漏洞测试 Builder测试过程 1 端口扫描 nmap -sC -sV 10.129.230.2202 测试思路 系统开启了22和8080端口&#xff0c;22端口无账号密码&#xff0c;测试方向主要从8080的jenkins服务开始测试。 在测试开源系统时&#xff0c;可以下载源代码或本地…
阅读更多...
基于springboot的一站式家装服务管理系统

基于springboot的一站式家装服务管理系统

文章目录 项目介绍主要功能截图：部分代码展示设计总结项目获取方式🍅 作者主页：超级无敌暴龙战士塔塔开 🍅 简介：Java领域优质创作者🏆、 简历模板、学习资料、面试题库【关注我，都给你】 🍅文末获取源码联系🍅 项目介绍 基于springboot的一站式家装服务管理系统…
阅读更多...
766. 托普利茨矩阵

766. 托普利茨矩阵

给你一个 m x n 的矩阵 matrix 。如果这个矩阵是托普利茨矩阵&#xff0c;返回 true &#xff1b;否则&#xff0c;返回 false 。 如果矩阵上每一条由左上到右下的对角线上的元素都相同&#xff0c;那么这个矩阵是 托普利茨矩阵 。 示例 1&#xff1a; 输入&#xff1a;matr…
阅读更多...
3、jvm基础知识（三）

3、jvm基础知识（三）

如何判断堆上的对象没有被引用&#xff1f; 常见的有两种判断方法&#xff1a;引用计数法和可达性分析法。 引用计数法会为每个对象维护一个引用计数器&#xff0c;当对象被引用时加1&#xff0c;取消引用时减1。 引用计数法的优点是实现简单&#xff0c;缺点有两点&#xff1…
阅读更多...
电机选型与直线模组（简化版）

电机选型与直线模组（简化版）

2-1电机分类和普通电机选型 1、电机及其原理的基本认知&#xff1a; 电动机我们平时简称电机&#xff0c;大部分情况下指的都是通电后&#xff0c;输出轴会旋转运动的设备。即将电能转换为机械能&#xff0c;供我们使用以驱动机构运行。电机简单的讲&#xff0c;其实就是一个…
阅读更多...
ubuntu 一键安装python3.10和pip3.10

ubuntu 一键安装python3.10和pip3.10

在Ubuntu系统上&#xff0c;您可以通过添加官方Python PPA来安装Python 3.10。以下是通过命令行安装Python 3.10和pip的步骤&#xff1a; 打开终端。 添加deadsnakes PPA&#xff08;Personal Package Archive&#xff09;&#xff1a; sudo add-apt-repository ppa:deadsna…
阅读更多...
梨花带雨网页音乐播放器二开优化修复美化版全开源版本源码

梨花带雨网页音乐播放器二开优化修复美化版全开源版本源码

源码简介 最新梨花带雨网页音乐播放器二开优化修复美化版全开源版本源码下载 梨花带雨播放器基于thinkphp6开发的XPlayerHTML5网页播放器前台控制面板,支持多音乐平台音乐解析。二开内容&#xff1a;修复播放器接口问题&#xff0c;把接口本地化&#xff0c;但是集成外链播放器…
阅读更多...
【前端推荐】node.js有哪些命令

【前端推荐】node.js有哪些命令

Node.js是一个基于Chrome V8引擎的JavaScript运行环境&#xff0c;它允许开发者在服务器端运行JavaScript代码。Node.js拥有丰富的模块和工具&#xff0c;通过命令行工具&#xff0c;可以执行各种任务来简化开发流程。以下是一些常用的Node.js命令&#xff1a; 1. node 这是最…
阅读更多...
x264 arm64汇编分析 quant8x8_neon分析

x264 arm64汇编分析 quant8x8_neon分析

一 C语言实现 #define QUANT_ONE( coef, mf, f ) \ { \ if( (coef) > 0 ) \ (coef) (f (coef)) * (mf) >> 16; \ else \ (coef) - ((f - (coef)) * (mf) >> 16); \ nz | (coef); \ } static int quant_8x8( dctcoef dct[64], udctcoef mf[64], udctcoef bias…
阅读更多...
【Qt】常用控件（输入类）

【Qt】常用控件（输入类）

目录 一、Line Edit二、Text Edit三、ComBo四、DateTimeEdit五、Slider 一、Line Edit QLineEdit 用来表示单行输入框&#xff0c;可以输入一段文本&#xff0c;但是不能换行。 属性说明test输入框中的文本inputMask输入内容格式约束maxLength最大长度frame是否添加边框echoM…
阅读更多...
最新文章

Copyright @ 2022~2023