南京网站建设 ww,如何做x响应式网站,如何才能做好品牌网站建设,网站建设在国外文章目录 一、虚拟主机设计二、实现虚拟主机1, 创建 VirtualHost 类2, VirtualHost() 构造方法3, exchangeDeclare() 创建交换机4, exchageDelete() 删除交换机5, queueDeclare() 创建队列6, queueDelete() 删除队列7, queueBind() 创建绑定8, queueUnBind() 删除绑定9, basicP… 文章目录 一、虚拟主机设计二、实现虚拟主机1, 创建 VirtualHost 类2, VirtualHost() 构造方法3, exchangeDeclare() 创建交换机4, exchageDelete() 删除交换机5, queueDeclare() 创建队列6, queueDelete() 删除队列7, queueBind() 创建绑定8, queueUnBind() 删除绑定9, basicPublish() 发布消息10, basicSubscribe() 订阅消息11, basicAck() 确认应答 三、实现交换机的转发/绑定规则1, 设计 bindingKey 和 routingKey2, checkBindingKey() 检查 bindingKey是否合法3, checkRoutingKey() 检查 routingKey 是否合法4, route() 判定转发规则5, routeFanout() 规定扇出交换机转发规则6, routeTopic() 规定主题交换机转发规则 四、小结 创建 Spring Boot 项目, Spring Boot 2 系列版本, Java 8 , 引入 MyBatis, Lombok 依赖 提示是正在努力进步的小菜鸟一只如有大佬发现文章欠佳之处欢迎批评指点~ 废话不多说直接上干货
整体目录结构 :
本文主要实现 server 包中的 VirtualHost 类 一、虚拟主机设计
到本篇为止, 内存和硬盘的数据都已经组织完成. 接下来使⽤ “虚拟主机” 这个概念, 把这两部分的数据也串起来.并且实现⼀些 MQ 的核心 API
回顾 BrokerServer (中间人服务器) 中的核心概念 : 声明 在 RabbitMQ 中, 虚拟主机是可以随意创建/删除的. 此处为了实现简单, 并没有实现虚拟主机的管理. 因此我们默认就只有⼀个虚拟主机的存在. 但是在数据结构的设计上我们预留了对于多虚拟主机的管理 虚拟主机存在的目的, 就是为了隔离, 隔离不同业务线上的数据, 所有此处要考虑的是: 交换机和队列从属于虚拟主机中, 如何把不同虚拟主机中的交换机和队列区分开? 此处使用一个简单粗暴的方式, 我们设定交换机和队列的唯一身份标识加一个前缀, 这个前缀就是虚拟主机的唯一标识, 当交换机和队列区分开之后, 绑定和队列中的消息自然就区分开了
虚拟主机中需要的属性和方法 : consumerManager 这个对象在下文介绍 二、实现虚拟主机
1, 创建 VirtualHost 类
virtualHostName 是虚拟主机的唯一身份标识虚拟主机要整合硬盘上的数据(数据库, 文件)和内存上的数据, 统一管理和使用, 所以引入 memoryDataCenter 和 diskDataCenter 这两个成员属性router 用来定义交换机和队列之间的匹配规则和转发规则consumerManager 是虚拟主机实现和消费者相关的 API 时的辅助对象, 这个对象中包含和消费者相关的 API, 仅供内部调用, 不对外暴露exchangeLock 和 queueLock 是后续实现 API 时保证线程安全的锁对象
public class VirtualHost {private String virtualHostName;private MemoryDataCenter memoryDataCenter;private DiskDataCenter diskDataCenter;private Router router;private ConsumerManager consumerManager; private final Object exchangeLock new Object();private final Object queueLock new Object();
}所以在前几篇文章中介绍的, 文件管理和内存管理, 都是在某个虚拟主机中的, 如果有多个虚拟主机, 每个虚拟主机中都有对应的文件和内存 这篇文章介绍的文件管理 这篇文章介绍的内存管理 2, VirtualHost() 构造方法
在构造方法中实现对刚才定义的成员属性的初始化, 对数据库文件的初始化
并且有可能本次不是第一次启动, 而是重启, 就需要恢复硬盘上的数据到内存中 public VirtualHost(String virtualHostName) {this.virtualHostName virtualHostName;this.memoryDataCenter new MemoryDataCenter();this.diskDataCenter new DiskDataCenter();this.router new Router();this.consumerManager new ConsumerManager(this);diskDataCenter.init();try {memoryDataCenter.recover(diskDataCenter);} catch (MQException | IOException e) {System.out.println([VirtualHost] 内存恢复数据失败);e.printStackTrace();}}另外提供一些 getter() public MemoryDataCenter getMemoryDataCenter() {return memoryDataCenter;}public DiskDataCenter getDiskDataCenter () {return diskDataCenter;}public String getVirtualHostName() {return virtualHostName;}3, exchangeDeclare() 创建交换机 此处包括下文的核心 API 中的参数都是参考了 RabbitMQ, 并在 这篇文章 中介绍了创建 Exchange, Queue, Binding, Message等核心类时, 已经说明了部分属性本项目中暂不实现 先判断交换机是否已经存在, 如果不存在则创建, 如果存在也不会抛异常, 直接返回即可 根据参数中的 durable 判断是否要写入硬盘 /*** 创建交换机* param exchangeName 名称(唯一标识)* param exchangeTypeEnum 类型* param durable 是否持久化存储* param autoDelete 是否自动删除* param arguments 配置参数* return 已存在返回 true, 不存在则创建*/public boolean exchangeDeclare(String exchangeName,ExchangeTypeEnum exchangeTypeEnum,boolean durable,boolean autoDelete,MapString, Object arguments) {exchangeName virtualHostName - exchangeName;try {synchronized (exchangeLock) {// 1, 查询是否已经存在交换机Exchange exchangeExists memoryDataCenter.getExchange(exchangeName);if (exchangeExists ! null) {System.out.println([VirtualHost.exchangeDeclare()] exchangeName exchangeName 的交换机已存在, 创建失败);return true;}// 2, 创建交换机Exchange exchange new Exchange();exchange.setName(exchangeName);exchange.setType(exchangeTypeEnum);exchange.setDurable(durable);exchange.setAutoDelete(autoDelete);exchange.setArguments(arguments);// 3, 写入硬盘if (durable) {diskDataCenter.insertExchange(exchange);}// 4, 写入内存memoryDataCenter.addExchange(exchange);System.out.println([VirtualHost.exchangeDeclare()] exchangeName exchangeName 的交换机创建成功);return true;}} catch (Exception e) {System.out.println([VirtualHost.exchangeDeclare()] exchangeName exchangeName 的交换机创建失败);e.printStackTrace();return false;}}按照先写入硬盘, 再写入内存的顺序编写代码, 因为写硬盘失败概率更⼤, 如果硬盘写失败了, 也就不必写内存了 4, exchageDelete() 删除交换机
先使用交换机唯一标识查找交换机, 该交换机如果不存在, 就无法删除
如果该交换机是持久化存储的, 则先删除硬盘, 再删除内存 /*** 删除交换机* param exchangeName 唯一标识* return true/false*/public boolean exchangeDelete(String exchangeName) throws MQException {exchangeName virtualHostName - exchangeName;try {synchronized (exchangeLock) {// 1, 查找该交换机Exchange exchangeExists memoryDataCenter.getExchange(exchangeName);if (exchangeExists null) {throw new MQException(VirtualHost.exchangeDelete() exchangeName exchangeName 的交换机不存在, 删除失败);}// 2, 硬盘删除if (exchangeExists.isDurable()) {diskDataCenter.deleteExchange(exchangeName);}// 3, 内存删除memoryDataCenter.removeExchange(exchangeName);System.out.println(VirtualHost.exchangeDelete() exchangeName exchangeName 的交换机删除成功);return true;}} catch (MQException e) {System.out.println(VirtualHost.exchangeDelete() exchangeName exchangeName 的交换机删除失败);e.printStackTrace();return false;}}5, queueDeclare() 创建队列 /*** 创建队列* param queueName 唯一标识* param durable 是否持久化* param exclusive 是否被独占* param autoDelete 是否自动删除* param arguments 额外参数* return true/false*/public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,MapString, Object arguments) {queueName virtualHostName - queueName;try {synchronized (queueLock) {// 1, 查找是否存在MessageQueue queueExists memoryDataCenter.getQueue(queueName);if (queueExists ! null) {System.out.println([VirtualHost.queueDeclare()] queueName queueName 的队列已存在, 创建失败);return true;}// 2, 创建队列MessageQueue queue new MessageQueue();queue.setName(queueName);queue.setDurable(durable);queue.setExclusive(exclusive);queue.setAutoDelete(autoDelete);queue.setArguments(arguments);// 3, 硬盘存储if (durable) {diskDataCenter.insertQueue(queue);}// 4, 内存存储memoryDataCenter.addQueue(queue);System.out.println([VirtualHost.queueDeclare()] queueName queueName 的队列创建成功);return true;}} catch (Exception e) {System.out.println([VirtualHost.queueDeclare()] queueName queueName 的队列创建失败);e.printStackTrace();return false;}}6, queueDelete() 删除队列 /*** 删除队列* param queueName 唯一标识* return true/false*/public boolean queueDelete(String queueName) {queueName virtualHostName - queueName;try {synchronized (queueLock) {MessageQueue queueExists memoryDataCenter.getQueue(queueName);// 1, 检查是否存在if (queueExists null) {throw new MQException(VirtualHost.queueDelete() queueName queueName 的队列已经存在, 删除失败);}// 2, 硬盘删除if (queueExists.isDurable()) {diskDataCenter.deleteQueue(queueName);}// 3, 内存删除memoryDataCenter.removeQueue(queueName);System.out.println(VirtualHost.queueDelete() queueName queueName 的队列删除成功);return true;}} catch (Exception e) {System.out.println(VirtualHost.queueDelete() queueName queueName 的队列删除失败);e.printStackTrace();return false;}}7, queueBind() 创建绑定
先检查交换机和队列的绑定是否存在, 如果存在, 则不创建绑定, 也不抛异常, 直接返回即可检查 bindingKey 是否合法(后续介绍判定规则)检查指定的交换机和队列是否存在, 如果不存在, 自然不能创建绑定, 应该抛异常 public boolean queueBind(String exchangeName, String queueName, String bindingKey) {exchangeName virtualHostName - exchangeName;queueName virtualHostName - queueName;try {synchronized (exchangeLock) {synchronized (queueLock) {// 1, 检查绑定是否存在Binding bindingExists memoryDataCenter.getBinding(exchangeName, queueName);if (bindingExists ! null) {System.out.println([VirtualHost.queueBind()] exchangeName exchangeName , queueName queueName 的绑定已经存在, 无需重复创建);return true; // 这里是否应该允许true??这里是否应该允许true??这里是否应该允许true??这里是否应该允许true??这里是否应该允许true??这里是否应该允许true??}// 2, 检查routingKey是否合法if (!router.checkBindingKey(bindingKey)) {throw new MQException([VirtualHost.queueBind()] bindingKey bindingKey 不合法, 绑定创建失败);}// 3, 创建binding对象Binding binding new Binding();binding.setExchangeName(exchangeName);binding.setQueueName(queueName);binding.setBindingKey(bindingKey);// 4, 检查交换机/队列是否存在Exchange exchangeExists memoryDataCenter.getExchange(exchangeName);MessageQueue queueExists memoryDataCenter.getQueue(queueName);if(exchangeExists null) {throw new MQException([VirtualHost.queueBind()] exchangeName exchangeName 的队列不存在, 绑定创建失败);}if(queueExists null){throw new MQException([VirtualHost.queueBind()] queueName queueName 的队列不存在, 绑定创建失败);}// 5, 写入硬盘if (exchangeExists.isDurable() queueExists.isDurable()) {diskDataCenter.insertBinding(binding);}// 6, 写入内存memoryDataCenter.addBinding(binding);System.out.println([VirtualHost.queueBind()] exchangeName exchangeName , queueName queueName 的绑定创建成功);return true;}}} catch (Exception e) {System.out.println([VirtualHost.queueBind()] exchangeName exchangeName , queueName queueName 的绑定创建失败);e.printStackTrace();return false;}}8, queueUnBind() 删除绑定
先检查绑定是否存在, 如果不存在, 自然不能删除, 应该抛异常 public boolean queueUnBind(String exchangeName, String queueName) {exchangeName virtualHostName - exchangeName;queueName virtualHostName - queueName;try {synchronized (exchangeLock) {synchronized (queueLock) {// 1, 检查 binding 是否存在Binding bindingExists memoryDataCenter.getBinding(exchangeName, queueName);if (bindingExists null) {throw new MQException([VirtualHost.queueUnBind()] exchangeName exchangeName , queueName queueName 的绑定不存在, 删除失败);}// 2, 硬盘删除diskDataCenter.deleteBinding(bindingExists);// 3, 内存删除memoryDataCenter.removeBinding(bindingExists);System.out.println([VirtualHost.queueUnBind()] exchangeName exchangeName , queueName queueName 的绑定删除成功);return true;}}} catch (Exception e) {System.out.println([VirtualHost.queueUnBind()] exchangeName exchangeName , queueName queueName 的绑定删除失败);e.printStackTrace();return false;}}9, basicPublish() 发布消息
在创建绑定的时候判定了 bindingKey 是否合法, 在此处发布消息的时候要判定 routingKey 是否合法(这个方法下面介绍)检查交换机是否存在如果是直接交换机, 直接把 routingKey 作为队列名, 找到该队列如果是扇出/主题交换机, 要先判定 bindingKey 和 routingKey 是否匹配, 如果匹配才能转发(这个方法下面介绍) public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body) {exchangeName virtualHostName - exchangeName;try {// 1, 检查routingKey是否合法if (!router.checkRoutingKey(routingKey)) {throw new MQException([VirtualHost.basicPublish()] routingKey routingKey 不合法, 消息发布失败);}// 2, 查找交换机是否存在Exchange exchangeExists memoryDataCenter.getExchange(exchangeName);if (exchangeExists null) {throw new MQException([VirtualHost.basicPublish()] exchangeName exchangeName 的交换机不存在, 消息发布失败);}// 3, 构造消息对象Message message Message.createMessage(routingKey, basicProperties, body);// 4, 判定交换机类型if (exchangeExists.getType() ExchangeTypeEnum.DIRECT) {// 如果是直接交换机, routingKey就作为队列名String queueName virtualHostName - routingKey;// 判定队列是否存在MessageQueue queueExists memoryDataCenter.getQueue(queueName);if (queueExists null) {throw new MQException([VirtualHost.basicPublish()] queueName queueName 的队列不存在, 消息发布失败);}sendMessage(queueExists, message);} else {ConcurrentHashMapString, Binding bindingsMap memoryDataCenter.getBindings(exchangeName);for (Map.EntryString, Binding entry : bindingsMap.entrySet()) {Binding binding entry.getValue();// 判定队列是否存在, 以及 bindingKey 和 routingKey 是否匹配MessageQueue queueExists memoryDataCenter.getQueue(binding.getQueueName());if (queueExists null || !router.route(exchangeExists.getType(), binding, message)) {continue;}sendMessage(queueExists, message);}}return true;} catch (MQException e) {System.out.println([VirtualHost.basicPublish()] 消息发布失败);e.printStackTrace();return false;}}首先队列为持久化存储, 并且该消息是持久化存储, 才能写入硬盘生产者发布了消息, 此时应该提醒消费者消费消息(这个方法后续实现) private void sendMessage(MessageQueue queue, Message message) {try {// 1, 写入硬盘if ( queue.isDurable() message.getDeliverMode() 2 ) {diskDataCenter.sendMessage(queue, message);}// 2, 写入内存memoryDataCenter.sendMessage(queue, message);// 3, 给消费者推送消息(message--queue, queueName--tokenQueue)consumerManager.notifyConsumer(queue.getName());} catch (InterruptedException | IOException | MQException e) {System.out.println([VirtualHost.sendMessage()] 消息发送失败);e.printStackTrace();}}10, basicSubscribe() 订阅消息
这个方法涉及到 ConsumerManager 这个类的实现, 下篇文章再介绍 ConsumerManager
这个方法只需要做一步: 添加一个消费者, 后续的逻辑(服务器收到消息后就转发给消费者)交给 ConsumerManager 处理 /*** 添加一个指定队列的消费者, 来订阅消息* 队列中有消息了就推送给订阅了该队列的消费者(订阅者)* param consumerTag 消费者唯一身份标识* param queueName 队列唯一身份标识* param autoAck 是否自动应答* param consumable 实现把消息推送给订阅者的接口*/public boolean basicSubscribe(String consumerTag,String queueName,boolean autoAck,Consumable consumable) {queueName virtualHostName - queueName;try {consumerManager.addConsumer(consumerTag, queueName, autoAck, consumable);System.out.println([VirtualHost.basicSubscribe()] consumerTag consumerTag 添加消费者成功);return true;} catch (MQException e) {System.out.println([VirtualHost.basicSubscribe()] consumerTag consumerTag 添加消费者失败);e.printStackTrace();return false;}}11, basicAck() 确认应答
这个方法用于, 服务器给消费者推送消息之后, 如果消费者选择手动应答, 就应该主动的调用服务器的这个方法
先查询队列和消息是否都存在如果都存在, 分别在硬盘和内存中, 删除该消息的即可 public boolean basicAck(String queueName, String messageId) {queueName virtualHostName - queueName;try {// 1, 查找队列和消息MessageQueue queueExists memoryDataCenter.getQueue(queueName);if (queueExists null) {throw new MQException([VirtualHost.basicAck()] queueName queueName 的交换机不存在, 确认应答失败);}Message messageExists memoryDataCenter.getMessage(messageId);if (messageExists null) {throw new MQException([VirtualHost.basicAck()] messageId messageId 的消息不存在, 确认应答失败);}// 2, 硬盘删除if (queueExists.isDurable() messageExists.getDeliverMode() 2) {diskDataCenter.deleteMessage(queueExists, messageExists);}// 3, 内存删除memoryDataCenter.removeMessage(messageId);memoryDataCenter.removeMessageNotAck(queueName, messageId);System.out.println([VirtualHost.basicAck()] queueName queueName , messageId messageId 的确认应答成功);return true;} catch (MQException | IOException e) {System.out.println([VirtualHost.basicAck()] queueName queueName , messageId messageId 的确认应答失败);e.printStackTrace();return false;}}三、实现交换机的转发/绑定规则 在 server.core.Router 类中编写代码, 这篇文章 介绍了核心类的实现, 当时没有实现这一块的内容 1, 设计 bindingKey 和 routingKey
⼀个 routingKey 是由数字, 字⺟, 下划线构成的, 并且可以使⽤ . 分成若⼲部分. 形如 aaa.bbb.ccc ⼀个 bindingKey 是由数字, 字⺟, 下划线构成的, 并且使⽤ . 分成若⼲部分. 另外, ⽀持 * 和 # 两种通配符. (*, # 只能作为 . 切分出来的独⽴部分, 不能和其他数字字⺟混⽤, ⽐如 a.*.b 是合法的, a.*a.b 是不合法的).
其中 * 可以匹配任意⼀个单词.其中 # 可以匹配任意零个或者多个单词.
bindingKey 为 a.*.b, 可以匹配 routingKey 为 a.a.b 和 a.b.b 和 a.aaa.b bindingKey 为 a.#.b, 可以匹配 routingKey 为 a.a.b 和 a.b.b 和 a.aaa.b 和 a.aa.bb.b 和 a.b 2, checkBindingKey() 检查 bindingKey是否合法
bindngKey可以为(空串), 如果交换机类型为直接 / 扇出, bindingKey 用不上, 参数传即可
允许是空字符串数字字⺟下划线构成可以包含通配符# 不能连续出现# 和 *不能相邻 public boolean checkBindingKey(String bindingKey) {if (bindingKey.length() 0) {return true;}for (int i 0; i bindingKey.length(); i) {char ch bindingKey.charAt(i);if ((ch A ch Z) || (ch a ch z) || (ch 0 ch 9)|| (ch _ || ch . || ch * || ch #)) {continue;}return false;}String[] bindingKeyFragments bindingKey.split(\\.);for (String fragment : bindingKeyFragments) {if (fragment.length() 1 (fragment.contains(*) || fragment.contains(#))) {return false;}}for (int i 0; i bindingKeyFragments.length - 1; i) {// 连续两个 ##if (bindingKeyFragments[i].equals(#) bindingKeyFragments[i 1].equals(#)) {return false;}// # 连着 *if (bindingKeyFragments[i].equals(#) bindingKeyFragments[i 1].equals(*)) {return false;}// * 连着 #if (bindingKeyFragments[i].equals(*) bindingKeyFragments[i 1].equals(#)) {return false;}}return true;}3, checkRoutingKey() 检查 routingKey 是否合法
如果是扇出交换机, routingKey 用不上, 设置为即可 public boolean checkRoutingKey(String routingKey) {if (routingKey.length() 0) {return true;}for (int i 0; i routingKey.length(); i) {char ch routingKey.charAt(i);if ((ch A ch Z) || (ch a ch z) || (ch 0 ch 9)|| (ch _ || ch . )) {continue;}return false;}return true;}4, route() 判定转发规则
直接交换机没有转发这一说, 在上述 basicPublish() 这个方法里已经单独处理过了 public boolean route(ExchangeTypeEnum exchangeTypeEnum, Binding binding, Message message) throws MQException {if (exchangeTypeEnum ExchangeTypeEnum.FANOUT) {return routeFanout(binding, message);} else if (exchangeTypeEnum ExchangeTypeEnum.TOPIC) {return routeTopic(binding, message);}else {throw new MQException([Router.route()] 非法的交换机类型);}}5, routeFanout() 规定扇出交换机转发规则
没有转发规则, 只要是绑定了的队列都能转发, 这里单拎出来一个方法是为了代码风格统一 public boolean routeFanout(Binding binding, Message message) {return true;}6, routeTopic() 规定主题交换机转发规则
这个方法就是用来实现判定 routingKey 和 bindingKey 是否匹配了 public boolean routeTopic(Binding binding, Message message) {String[] bindingKeyFragments binding.getBindingKey().split(\\.);String[] routingKeyFragments message.getRoutingKey().split(\\.);int i 0;int j 0;while (i bindingKeyFragments.length j routingKeyFragments.length) {// 遇到 * 只能匹配一个字符if (bindingKeyFragments[i].equals(*)) {i;j;continue;}// 遇到 # 就找下一个片段if (bindingKeyFragments[i].equals(#)) {i;if (i bindingKeyFragments.length) {return true;}// 说明#后面还有片段, 让 j 寻找这个片段int nextMatchIndex findNextMatchIndex(bindingKeyFragments[i], j, routingKeyFragments);if (nextMatchIndex -1) {return false;}j nextMatchIndex;i;j;continue;}if (!bindingKeyFragments[i].equals(routingKeyFragments[j])) {return false;}i;j;}return i bindingKeyFragments.length j routingKeyFragments.length;}public int findNextMatchIndex(String cur, int j, String[] routingKeyFragments) {while (j routingKeyFragments.length) {if (routingKeyFragments[j].equals(cur)) {return j;}j;}return -1;}四、小结
本篇主要实现了虚拟主机 虚拟主机的作用是为了隔离不同业务线的数据, 本项目暂时只支持一个虚拟主机
虚拟主机把硬盘(数据库文件)和内存这两个模块的数据管理整合在一起, 并且封装了一系列核心 API, 供上层( BrokerServer )调用在server.core 包下实现了 Router 类, 这个类主要是在生产者发布消息时, 用来检查 bindingKey 和routingKey 是否合法, routingKey 和 bindingKey 是否匹配等, 为了实现服务器支持三种不同的交换机转发模式 这些核心 API 基本上都实现了, 还有 basicSubscribe() 这个 API 没有完全实现, 在下篇文章会介绍 ConsumerManager 类, 用来实现和消费者消费消息相关的逻辑
