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

RabbitMQ的部分模式

news 2025/9/11 16:58:44

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.lryc.cn/news/330186.html

相关文章：

  • 提取单选框的值，并通过ajax传值到后台
  • Django创建多app应用
  • 如何反反爬虫
  • wireshark抓包之DNS协议
  • 升级到 Java 21 是值得的
  • C# 多线程
  • 快速安装sudachipy日语包
  • 蓝桥杯刷题day13——乘飞机【算法赛】
  • 大模型量化技术-BitsAndBytes
  • EasyExcel 复杂表头的导出（动态表头和静态表头）
  • centos7 fatal error: curl/curl.h: No such file or directory
  • 【Linux】自定义协议+序列化+反序列化
  • 常见故障排查和优化
  • 选择华为HCIE培训机构有哪些注意事项
  • python怎么处理txt
  • SAMRTFORMS 转换PDF 发送邮件
  • 探讨在大数据体系中API的通信机制与工作原理
  • 算法打卡day23
  • 每天五分钟深度学习：神经网络和深度学习有什么样的关系？
  • 基于PSO优化的CNN-LSTM-Attention的时间序列回归预测matlab仿真
  • 物联网监控可视化是什么？部署物联网监控可视化大屏有什么作用？
  • 设计一个Rust线程安全栈结构 Stack＜T＞
  • Docker Desktop 在 Windows 上的安装和使用
  • 2024年最受欢迎的 19 个 VS Code 主题排行榜
  • 突破编程_C++_网络编程（OSI 七层模型（物理层与数据链路层））
  • Spring boot如何使用redis缓存
  • 红蓝色WordPress外贸建站模板
  • python爬虫----了解爬虫（十一天）
  • 碳素光线疗法与宠物健康
  • 展锐平台camera添加底层水印