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

2024.2.29 模拟实现 RabbitMQ —— 项目展示

news 2025/9/11 19:07:20

目录

项目介绍

核心功能

核心技术

演示直接交换机

演示扇出交换机

演示主题交换机

项目介绍

  • 此处我们模拟 RabbitMQ 实现了一个消息队列服务器

核心功能

  • 提供了 虚拟主机、交换机、队列、绑定、消息 概念的管理
  • 九大核心 API 创建队列、销毁队列、创建交换机、销毁交换机、创建绑定、解除绑定、发布消息、订阅消息、确认消息
  • 实现了三种典型的消息转换方式 直接交换机（Direct）、扇出交换机（Fanout）、主题交换机（Topic）
  • 交换机、队列、绑定 使用 SQLite 数据库持久化，消息 使用文件持久化
  • 基于 TCP + 自定义应用层协议 实现生产者/消费者和 Broker Server 之间的交互工作

核心技术

  • Spring Boot / MyBatis / Lombok
  • SQLite
  • TCP
  • 关于该项目的需求分析，可点击下方链接跳转

模拟实现 RabbitMQ —— 需求分析

  • 关于该项目的核心类，可点击下方链接跳转

模拟实现 RabbitMQ —— 实现核心类

  • 关于该项目的数据库操作，可点击下方链接跳转

模拟实现 RabbitMQ —— 数据库操作

  • 关于该项目的消息持久化，可点击下方链接跳转

模拟实现 RabbitMQ —— 消息持久化

  • 关于该项目的内存数据管理，可点击下方链接跳转

模拟实现 RabbitMQ —— 内存数据管理

  • 关于该项目的虚拟机设计，可点击下方链接跳转

模拟实现 RabbitMQ —— 虚拟主机设计

  • 关于该项目的交换机转发规则，可点击下方链接跳转

模拟实现 RabbitMQ —— 实现转发规则

  • 关于该项目的消费逻辑，可点击下方链接跳转

模拟实现 RabbitMQ —— 实现消费消息逻辑

  • 关于该项目网络通信设计，可点击下方链接跳转

模拟实现 RabbitMQ —— 网络通信设计（服务器）

模拟实现 RabbitMQ —— 网络通信设计（客户端）

演示直接交换机

  • 简单写一个 demo 模拟 跨主机的生产者消费者模型
  • 此处为了方便，就在本机演示

  • 此处我们创建的交换机类型为 直接交换机

1、在 Spring Boot 项目的启动类中创建 Broker Server，绑定端口并启动！

@SpringBootApplication
public class DemoApplication {public static ConfigurableApplicationContext context;public static void main(String[] args) throws IOException {context = SpringApplication.run(DemoApplication.class, args);BrokerServer brokerServer = new BrokerServer(9090);brokerServer.start();}
}

2、编写生产者代码

/*
* 这个类用来表示一个生产着
* 通常这是一个单独的服务器程序
* */
public class DemoProducer {public static void main(String[] args) throws IOException, InterruptedException {System.out.println("启动生产者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//        创建交换机和队列channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);channel.queueDeclare("testQueue",true,false,false,null);//        创建一个消息并发送byte[] body = "hello".getBytes();boolean ok = channel.basicPublish("testExchange","testQueue",null,body);System.out.println("消息投递完成！ ok = " + ok);Thread.sleep(500);channel.close();connection.close();}
}

3、编写消费者代码

/*
* 这个类表示一个消费者
* 通常这个类也应该是在一个独立的服务器中被执行
* */
public class DemoConsumer {public static void main(String[] args) throws IOException, InterruptedException, MqException {System.out.println("启动消费者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare("testExchange",ExchangeType.DIRECT,true,false,null);channel.queueDeclare("testQueue",true,false,false,null);channel.basicConsume("testQueue", true, new Consumer() {@Overridepublic void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {System.out.println("[消费数据] 开始！");System.out.println("consumerTag = " + consumerTag);System.out.println("basicProperties = " + basicProperties);String bodyString = new String(body,0,body.length);System.out.println("body = " + bodyString);System.out.println("[消费数据] 结束！");}});//        由于消费者也不知道生产者要生产多少消息，就在这里通过这个循环模拟一直等待消费while (true) {Thread.sleep(500);}}
}

4、启动 Spring Boot 项目（启动 Broker Server）

5、运行消费者代码

6、运行生产者代码

7、继续观察消费者的控制台

演示扇出交换机

  • 此处我们创建的交换机类型为 扇出交换机

 1、编写生产者代码

/** 这个类用来表示一个生产着* 通常这是一个单独的服务器程序* */
public class DemoProducer {public static void main(String[] args) throws IOException, InterruptedException {System.out.println("启动生产者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();
//        创建交换机channel.exchangeDeclare("testExchange", ExchangeType.FANOUT,true,false,null);
//        创建一个消息并发送byte[] body = "hello".getBytes();boolean ok = channel.basicPublish("testExchange","",null,body);System.out.println("消息投递完成！ ok = " + ok);Thread.sleep(500);channel.close();connection.close();}
}

3、编写消费者A 代码

/** 这个类表示一个消费者A* 通常这个类也应该是在一个独立的服务器中被执行* */
public class DemoConsumerA {public static void main(String[] args) throws IOException, InterruptedException, MqException {System.out.println("启动消费者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();
//        创建交换机channel.exchangeDeclare("testExchange",ExchangeType.FANOUT,true,false,null);
//        创建队列channel.queueDeclare("testQueue1",true,false,false,null);
//        设置绑定channel.queueBind("testQueue1","testExchange","");
//        订阅消息channel.basicConsume("testQueue1", true, new Consumer() {@Overridepublic void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {System.out.println("[testQueue1 消费数据] 开始！");System.out.println("consumerTag = " + consumerTag);System.out.println("basicProperties = " + basicProperties);String bodyString = new String(body,0,body.length);System.out.println("body = " + bodyString);System.out.println("[testQueue1 消费数据] 结束！");}});//        由于消费者也不知道生产者要生产多少消息，就在这里通过这个循环模拟一直等待消费while (true) {Thread.sleep(500);}}
}

4、编写消费者B 代码

/** 这个类表示一个消费者B* 通常这个类也应该是在一个独立的服务器中被执行* */
public class DemoConsumerB {public static void main(String[] args) throws IOException, InterruptedException, MqException {System.out.println("启动消费者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();
//        创建交换机channel.exchangeDeclare("testExchange", ExchangeType.FANOUT,true,false,null);
//        创建队列channel.queueDeclare("testQueue2",true,false,false,null);
//        设置绑定channel.queueBind("testQueue2","testExchange","");
//        订阅消息channel.basicConsume("testQueue2", true, new Consumer() {@Overridepublic void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {System.out.println("[testQueue1 消费数据] 开始！");System.out.println("consumerTag = " + consumerTag);System.out.println("basicProperties = " + basicProperties);String bodyString = new String(body,0,body.length);System.out.println("body = " + bodyString);System.out.println("[testQueue1 消费数据] 结束！");}});//        由于消费者也不知道生产者要生产多少消息，就在这里通过这个循环模拟一直等待消费while (true) {Thread.sleep(500);}}
}

5、启动 Spring Boot 项目（启动 Broker Server）

6、运行消费者A 代码

7、运行消费者B 代码

8、运行生产者代码

9、继续观察消费者A 的控制台

10、继续观察消费者B 的控制台

演示主题交换机

  • 此处我们创建的交换机为 主题交换机

 1、编写生产者代码

/** 这个类用来表示一个生产着* 通常这是一个单独的服务器程序* */
public class DemoProducer {public static void main(String[] args) throws IOException, InterruptedException {System.out.println("启动生产者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//        创建交换机和队列channel.exchangeDeclare("testExchange", ExchangeType.TOPIC,true,false,null);channel.queueDeclare("testQueue",true,false,false,null);//        创建消息A 并发送byte[] body = "hello".getBytes();boolean ok = channel.basicPublish("testExchange","ccc.aaa.bbb",null,body);System.out.println("消息投递完成！ ok = " + ok);//        创建消息B 并发送body = "hi".getBytes();ok = channel.basicPublish("testExchange","aaa.bbb",null,body);System.out.println("消息投递完成！ ok = " + ok);Thread.sleep(500);channel.close();connection.close();}
}

3、编写消费者代码

/** 这个类表示一个消费者* 通常这个类也应该是在一个独立的服务器中被执行* */
public class DemoConsumer {public static void main(String[] args) throws IOException, InterruptedException, MqException {System.out.println("启动消费者！");ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(9090);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();
//        创建交换机channel.exchangeDeclare("testExchange", ExchangeType.TOPIC,true,false,null);
//        创建队列channel.queueDeclare("testQueue",true,false,false,null);
//        设置绑定channel.queueBind("testQueue","testExchange","*.aaa.bbb");
//        订阅消息channel.basicConsume("testQueue", true, new Consumer() {@Overridepublic void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {System.out.println("[消费数据] 开始！");System.out.println("consumerTag = " + consumerTag);System.out.println("basicProperties = " + basicProperties);String bodyString = new String(body,0,body.length);System.out.println("body = " + bodyString);System.out.println("[消费数据] 结束！");}});
//        由于消费者也不知道生产者要生产多少消息，就在这里通过这个循环模拟一直等待消费while (true) {Thread.sleep(500);}}
}

4、启动 Spring Boot 项目（启动 Broker Server）

5、运行消费者代码

6、运行生产者代码

7、继续观察消费者的控制台

http://www.lryc.cn/news/307078.html

相关文章：

  • React htmlfor
  • 现代化数据架构升级：毫末智行自动驾驶如何应对年增20PB的数据规模挑战？
  • 理解Stable Diffusion、LoRA、Dreambooth、Hypernetworks、Textual Inversion、Checkpoint
  • spring boot3登录开发-2(1图形验证码接口实现)
  • 网络编程中的问题总结
  • 数据结构-关键路径
  • 进程间通信学习笔记（共享内存）
  • ChatGPT学习第三周
  • R语言混合效应（多水平/层次/嵌套）模型及贝叶斯实现技术应用
  • [C++]使用C++部署yolov9的tensorrt模型进行目标检测
  • eureka注册中心做了哪些事情/原理？
  • c语言经典测试题4
  • 设计模式(五)-观察者模式
  • MySQL-七种SQL优化
  • 针对Umi、React中遇到的 “xxxx”不能用作 JSX 组件 问题解决方案
  • 蓝桥杯备战刷题one（自用）
  • 设计模式(十) - 工厂方式模式
  • http协议基础与Apache的简单介绍
  • RabbitMQ的死信队列和延迟队列
  • PyQt 逻辑与界面分离
  • opengl播放3d pose 原地舞蹈脚来回飘动
  • Linux环境基础开发工具使用篇(三) git 与 gdb
  • mybatis----＞tx中weekend类
  • Shell echo、printf、test命令
  • 腾讯云主机Ubuntu22.04安装Odoo17
  • conda常用命令详解
  • Java面试——锁
  • Spring Boot与Netty：构建高性能的网络应用
  • ARMv8-AArch64 的异常处理模型详解之异常处理详解（同步异常和异步异常的分析和处理）
  • Elasticsearch：基于 Langchain 的 Elasticsearch Agent 对文档的搜索