浅谈如何自我实现一个消息队列服务器(2)——实现 broker server 服务器

文章目录

  • 一、实现 broker server 服务器
    • 1.1 创建一个SpringBoot项目
    • 1.2 创建Java类
  • 二、硬盘持久化存储 broker server 里的数据
    • 2.1 数据库存储
      • 2.1.1 浅谈SQLite
        • MyBatis
      • 2.1.2 如何使用SQLite
    • 2.2 使用DataBaseManager类封装数据库操作
    • 2.3 文件存储消息
      • 2.3.1 存储消息时,按队列维度展开
        • 2.3.1.1 逻辑删除
        • 2.3.1.2 GC(垃圾回收)
          • 2.3.1.2.1 使用复制算法实现GC
        • 2.3.1.3 文件存储消息的扩展内容(该功能暂未实现)
          • 2.3.1.3.1 文件拆分
          • 2.3.1.3.2 文件合并
    • 2.4 使用类MessageFileManager封装文件存储操作
  • 三、将broker server 里的数据存储在内存上

一、实现 broker server 服务器

上次我们提到生产者-消费者模型中,最重要的是实现broker server(服务器模块)。 为什么最重要的是实现 broker server 呢?broker server 作为一个通用服务器,它可以给任何具有数据存储、转发需求的客户端服务器使用。就像现在的数据库程序一样,不论开发者实现什么业务,只要该业务中涉及数据的持久化存储,就需要使用数据库。

broker server 中管理着许多重要的概念(虚拟主机、交换机、队列、消息、绑定),这些概念的数据组织是实现 broker server 的关键。我们 以 如何使用代码实现一个 broker server 为核心,对 broker server 的实现进行详细介绍。

1.1 创建一个SpringBoot项目

之前的博客有介绍过如何使用IDEA创建一个 SpringBoot 项目,可以点击此处链接浅谈如何创建一个SpringBoot项目进行学习。

创建好一个SpringBoot项目之后,我们在pom.xml文件中引入SQLite的依赖,
在这里插入图片描述

再在配置文件application.yaml中针对SQLite、MyBatis的使用进行配置。

在这里插入图片描述

1.2 创建Java类

我们为 broker server 的实现创建一个包mqServer(这只是一个包名,可以自己随意进行自定义):
在这里插入图片描述

再在mqServer包下创建一个core包,core包里存放交换机、队列、消息、绑定这些概念的实体类:
在这里插入图片描述
交换机类(Exchange):
在这里插入图片描述

/** 这个类表示一个交换机*/
public class Exchange {// 此处使用 name 来作为交换机的身份标识. (唯一的)private String name;// 交换机类型, DIRECT, FANOUT, TOPICprivate ExchangeType type = ExchangeType.DIRECT;// 该交换机是否要持久化存储. true 表示需要持久化; false 表示不必持久化.private boolean durable = false;// 如果当前交换机, 没人使用了, 就会自动被删除.// 这个属性暂时先列在这里, 后续的代码中并没有真的实现这个自动删除功能~~ (RabbitMQ 是有的)private boolean autoDelete = false;// arguments 表示的是创建交换机时指定的一些额外的参数选项. 后续代码中并没有真的实现对应的功能, 先列出来. (RabbitMQ 也是有的)// 为了把这个 arguments 存到数据库中, 就需要把 Map 转成 json 格式的字符串.private Map<String, Object> arguments = new HashMap<>();public String getName() {return name;}public void setName(String name) {this.name = name;}public ExchangeType getType() {return type;}public void setType(ExchangeType type) {this.type = type;}public boolean isDurable() {return durable;}public void setDurable(boolean durable) {this.durable = durable;}public boolean isAutoDelete() {return autoDelete;}public void setAutoDelete(boolean autoDelete) {this.autoDelete = autoDelete;}// 这里的 get set 用于和数据库交互使用.public String getArguments() {// 是把当前的 arguments 参数, 从 Map 转成 String (JSON)ObjectMapper objectMapper = new ObjectMapper();try {return objectMapper.writeValueAsString(arguments);} catch (JsonProcessingException e) {e.printStackTrace();}// 如果代码真异常了, 返回一个空的 json 字符串就 okreturn "{}";}// 这个方法, 是从数据库读数据之后, 构造 Exchange 对象, 会自动调用到public void setArguments(String argumentsJson) {// 把参数中的 argumentsJson 按照 JSON 格式解析, 转成// 上述的 Map 对象ObjectMapper objectMapper = new ObjectMapper();try {this.arguments = objectMapper.readValue(argumentsJson, new TypeReference<HashMap<String, Object>>() {});} catch (JsonProcessingException e) {e.printStackTrace();}}// 在这里针对 arguments, 再提供一组 getter setter , 用来去更方便的获取/设置这里的键值对.// 这一组在 java 代码内部使用 (比如测试的时候)public Object getArguments(String key) {return arguments.get(key);}public void setArguments(String key, Object value) {arguments.put(key, value);}public void setArguments(Map<String, Object> arguments) {this.arguments = arguments;}
}

交换机类型 类(ExchangeType):
这个类是个枚举类,作为Exchange类的成员变量。

public enum ExchangeType {DIRECT(0),FANOUT(1),TOPIC(2);private final int type;private ExchangeType(int type) {this.type = type;}public int getType() {return type;}
}

队列类(MSGQueue):
在这里插入图片描述

/** 这个类表示一个存储消息的队列* MSG => Message*/
public class MSGQueue {// 表示队列的身份标识.private String name;// 表示队列是否持久化, true 表示持久化保存, false 表示不持久化.private boolean durable = false;// 这个属性为 true, 表示这个队列只能被一个消费者使用(别人用不了). 如果为 false 则是大家都能使用// 这个 独占 功能, 也是先把字段列在这里, 具体的独占功能暂时先不实现.private boolean exclusive = false;// 为 true 表示没有人使用之后, 就自动删除. false 则是不会自动删除.// 这个 自动删除 功能, 也是先把字段列在这里, 具体的独占功能暂时先不实现.private boolean autoDelete = false;// 也是表示扩展参数. 当前也是先列在这里, 先暂时不实现private Map<String, Object> arguments = new HashMap<>();/*** 因为消息是存在队列里的,每个队列下面都有相关的两个消息文件,生产者生产出消息之后,将* 消息投递给了broker server 服务器,消费者通过服务器订阅消息,消费者订阅到消息之后* 消费者具体想要使用这些消息做些什么,我们不知道,这得由消费者自己决定,所以消费者使用回调函数* Consumer之后,使用里面的抽象方法handleDelivery自己决定自己想要将此消息用来做些什么(重写)*** 因此在MSGQueue类里面,我们需要定义一个具有ConsumerEnv类型的List,* 表示当前队列里的消息都有哪些消费者订阅了*/// 当前队列都有哪些消费者订阅了.private List<ConsumerEnv> consumerEnvList = new ArrayList<>();// 记录当前取到了第几个消费者. 方便实现轮询策略.
//    压根没懂这个 todoprivate AtomicInteger consumerSeq = new AtomicInteger(0);
//    添加一个新的订阅者public void addConsumerEnv(ConsumerEnv consumerEnv){synchronized (this){consumerEnvList.add(consumerEnv);}}//    删除一个订阅者//    挑选一个订阅者,用来处理当前的消息(按照轮询的方式)public ConsumerEnv chooseConsumer(){if(consumerEnvList.size() == 0){
//            当前队列没人订阅return null;}//        压根没懂这个 todoint index = consumerSeq.get() % consumerEnvList.size();consumerSeq.getAndIncrement();return consumerEnvList.get(index);}public String getName() {return name;}public void setName(String name) {this.name = name;}public boolean isDurable() {return durable;}public void setDurable(boolean durable) {this.durable = durable;}public boolean isExclusive() {return exclusive;}public void setExclusive(boolean exclusive) {this.exclusive = exclusive;}public boolean isAutoDelete() {return autoDelete;}public void setAutoDelete(boolean autoDelete) {this.autoDelete = autoDelete;}public String getArguments() {ObjectMapper objectMapper = new ObjectMapper();try {return objectMapper.writeValueAsString(arguments);} catch (JsonProcessingException e) {e.printStackTrace();}return "{}";}public void setArguments(String argumentsJson) {ObjectMapper objectMapper = new ObjectMapper();try {this.arguments = objectMapper.readValue(argumentsJson, new TypeReference<HashMap<String, Object>>() {});} catch (JsonProcessingException e) {e.printStackTrace();}}public Object getArguments(String key) {return arguments.get(key);}public void setArguments(String key, Object value) {arguments.put(key, value);}public void setArguments(Map<String, Object> arguments) {this.arguments = arguments;}
}

绑定类(Binding):
在这里插入图片描述

/** 表示队列和交换机之间的关联关系*/
public class Binding {private String exchangeName;private String queueName;// bindingKey 就是在出题, 要求领红包的人要画个 "桌子" 出来~~private String bindingKey;// Binding 这个东西, 依附于 Exchange 和 Queue 的!!!// 比如, 对于持久化来说, 如果 Exchange 和 Queue 任何一个都没有持久化,// 此时你针对 Binding 持久化是没有任何意义的public String getExchangeName() {return exchangeName;}public void setExchangeName(String exchangeName) {this.exchangeName = exchangeName;}public String getQueueName() {return queueName;}public void setQueueName(String queueName) {this.queueName = queueName;}public String getBindingKey() {return bindingKey;}public void setBindingKey(String bindingKey) {this.bindingKey = bindingKey;}
}

消息类(Message):
在这里插入图片描述

/** 表示一个要传递的消息*/
public class Message implements Serializable {// 这两个属性是 Message 最核心的部分.private BasicProperties basicProperties = new BasicProperties();private byte[] body;// 下面的属性则是辅助用的属性.// Message 后续会存储到文件中(如果持久化的话).// 一个文件中会存储很多的消息. 如何找到某个消息, 在文件中的具体位置呢?// 使用下列的两个偏移量来进行表示. [offsetBeg, offsetEnd)// 这俩属性并不需要被序列化保存到文件中~~ 此时消息一旦被写入文件之后, 所在的位置就固定了. 并不需要单独存储.// 这俩属性存在的目的, 主要就是为了让内存中的 Message 对象, 能够快速找到对应的硬盘上的 Message 的位置.private transient long offsetBeg = 0;  // 消息数据的开头距离文件开头的位置偏移(字节)private transient long offsetEnd = 0;  // 消息数据的结尾距离文件开头的位置偏移(字节)// 使用这个属性表示该消息在文件中是否是有效消息. (针对文件中的消息, 如果删除, 使用逻辑删除的方式)// 0x1 表示有效. 0x0 表示无效.private byte isValid = 0x1;// 创建一个工厂方法, 让工厂方法帮我们封装一下创建 Message 对象的过程.// 这个方法中创建的 Message 对象, 会自动生成唯一的 MessageId// 万一 routingKey 和 basicProperties 里的 routingKey 冲突, 以外面的为主.public static Message createMessageWithId(String routingKey, BasicProperties basicProperties, byte[] body) {Message message = new Message();if (basicProperties != null) {message.setBasicProperties(basicProperties);}// 此处生成的 MessageId 以 M- 作为前缀.message.setMessageId("M-" + UUID.randomUUID());message.setRoutingKey(routingKey);message.body = body;// 此处是把 body 和 basicProperties 先设置出来. 他俩是 Message 的核心内容.// 而 offsetBeg, offsetEnd, isValid, 则是消息持久化的时候才会用到. 在把消息写入文件之前再进行设定.// 此处只是在内存中创建一个 Message 对象.return message;}public String getMessageId() {return basicProperties.getMessageId();}public void setMessageId(String messageId) {basicProperties.setMessageId(messageId);}public String getRoutingKey() {return basicProperties.getRoutingKey();}public void setRoutingKey(String routingKey) {basicProperties.setRoutingKey(routingKey);}public int getDeliverMode() {return basicProperties.getDeliverMode();}public void setDeliverMode(int mode) {basicProperties.setDeliverMode(mode);}public BasicProperties getBasicProperties() {return basicProperties;}public void setBasicProperties(BasicProperties basicProperties) {this.basicProperties = basicProperties;}public byte[] getBody() {return body;}public void setBody(byte[] body) {this.body = body;}public long getOffsetBeg() {return offsetBeg;}public void setOffsetBeg(long offsetBeg) {this.offsetBeg = offsetBeg;}public long getOffsetEnd() {return offsetEnd;}public void setOffsetEnd(long offsetEnd) {this.offsetEnd = offsetEnd;}public byte getIsValid() {return isValid;}public void setIsValid(byte isValid) {this.isValid = isValid;}@Overridepublic String toString() {return "Message{" +"basicProperties=" + basicProperties +", body=" + Arrays.toString(body) +", offsetBeg=" + offsetBeg +", offsetEnd=" + offsetEnd +", isValid=" + isValid +'}';}
}

BasicProperties 类(是Message类的成员变量):

public class BasicProperties implements Serializable {// 消息的唯一身份标识. 此处为了保证 id 的唯一性, 使用 UUID 来作为 message idprivate String messageId;// 是一个消息上带有的内容, 和 bindingKey 做匹配.// 如果当前的交换机类型是 DIRECT, 此时 routingKey 就表示要转发的队列名.// 如果当前的交换机类型是 FANOUT, 此时 routingKey 无意义(不使用).// 如果当前的交换机类型是 TOPIC, 此时 routingKey 就要和 bindingKey 做匹配. 符合要求的才能转发给对应队列.private String routingKey;// 这个属性表示消息是否要持久化. 1 表示不持久化, 2 表示持久化. (RabbitMQ 就是这样搞的....)private int deliverMode = 1;// 其实针对 RabbitMQ 来说, BasicProperties 里面还有很多别的属性. 其他的属性暂时先不考虑了.public String getMessageId() {return messageId;}public void setMessageId(String messageId) {this.messageId = messageId;}public String getRoutingKey() {return routingKey;}public void setRoutingKey(String routingKey) {this.routingKey = routingKey;}public int getDeliverMode() {return deliverMode;}public void setDeliverMode(int deliverMode) {this.deliverMode = deliverMode;}@Overridepublic String toString() {return "BasicProperties{" +"messageId='" + messageId + '\'' +", routingKey='" + routingKey + '\'' +", deliverMode=" + deliverMode +'}';}
}

二、硬盘持久化存储 broker server 里的数据

上述我们创建了 交换机、队列、绑定、消息 的实体Java类,这些概念的数据需要在内存、硬盘上各存一份。所以此时我们首先来讨论 交换机(Exchange)、队列(MSGQueue)、消息(Message)、绑定(Binding) 他们以何种方式存储在硬盘上。 至于他们在内存中以什么样的数据结构进行存储,我们后续再讨论。

2.1 数据库存储

对于交换机、队列、绑定来说,由于他们的数据量没有消息多,同时他们还需经常性进行 增删改查 操作,因此他们使用数据库进行数据的持久化存储。一般我们都是使用MySQL进行数据存储,但是由于MySQL是个客户端-服务器结构的程序,它本身具有一定的重量。在这里,我们为了简便,采用轻量的SQLite数据库进行数据存储。

2.1.1 浅谈SQLite

可能很多同学一开始学习使用数据库进行数据存储时,接触到的就是SQL Server、MySQL、Oracle这些数据库(包括我寄自),没怎么听说过SQLite。

但SQLite的应用也很广泛。

SQLite很轻量,它是一个可执行文件exe,主要在一些性能不高的设备上使用,尤其是在移动端(手机、APP)和嵌入式设备(投影仪、冰箱、交换机、路由器…),Android系统就内置了SQLite。虽然SQLite 轻量,但其具有的功能不输MySQL,其sql语句使用起来与MySQL基本无异,也支持通过MyBatis这样的框架来使用。(学习过安卓开发课程的同学肯定知道SQLite)。

SQLite 是一个本地数据库,把数据存储在当前硬盘的某个指定文件中,其无法跨网络使用,这个数据库相当于直接操作本地的硬盘文件,其将每个数据库database抽取成一个单独文件。

MyBatis

在介绍如何使用SQLite之前,先了解一下如何使用MyBatis吧。因为我发现有些同学还不太理解如何使用MyBatis操作数据库,当然了,掌握了MyBatis的同学可以直接跳过此处内容科普。

MyBatis是Spring框架集成的一个操作数据库的框架,以前我们想要在项目中连接数据库来进行数据的存储,使用的是JDBC,但是使用JDBC将后端代码与数据库连接时,发现JDBC其代码量较大,代码重复率高,同时sql语句以及Java代码都杂揉在了一起,代码显得不太优雅。因此现在Java项目中主流使用MyBatis操作数据库。

使用MyBatis操作数据库的流程:
1、一般是在项目中创建一个名叫mapper的包,在该包下创建interface(interface的数量由自己项目的业务代码决定),描述有哪些方法要给Java代码使用。

2、创建对应的xml文件,通过xml文件来实现上述interface中的抽象方法。
在这里插入图片描述
如果觉得MyBatis这里讲得比较抽象,不理解,我再找个时间专门出一篇关于MyBatis详解的博客。

2.1.2 如何使用SQLite

1、在Java中使用SQLite,无需安装以及下载任何东西。直接使用Maven,将SQLite的依赖引进pom.xml文件中即可,此时Maven会自动从中央仓库加载SQLite的jar包和dll(动态库)。
在这里插入图片描述
SQLite的依赖:

<dependency><groupId>org.xerial</groupId><artifactId>sqlite-jdbc</artifactId><version>3.41.0.1</version></dependency>

2、我们还需要在项目的配置文件application.yaml中,针对SQLite、MyBatis进行配置:
在这里插入图片描述
此时,我们就可以在项目中使用数据库SQLite进行数据持久化存储了。

在将 交换机、队列、绑定 持久化存储在数据库时,有一个需要注意的点:交换机类、队列类,他们都有一个Map类型的成员变量arguments。
在这里插入图片描述
由于数据库存储数据时,不支持Map类型的数据,因此当我们需要将Map类型的arguments变量在数据库中存储时,Map类型的arguments在数据库中就需要使用字符串类型表示。
在这里插入图片描述
我们知道MyBatis操作数据库进行写数据操作时,是调用对象属性对应的getter方法,将从getter方法获取到的属性的值写入数据库中;数据库进行读数据操作时,是调用对象的对应属性的setter方法,将从数据库获取到的值赋值给对应属性。

那么针对arguments这个变量,其数据类型在对象中是Map类型,但在数据库中其数据类型是字符串,那么如果数据库想要进行写操作时,就需要通过arguments变量的getter方法获取到一个String(字符串)类型的arguments,此时获取到的arguments类型才会与数据库中的字符串类型arguments相匹配,才能正确的进行数据库的写操作。但是由于在对象中的arguments它的数据类型是Map类型,此时对象中提供的getArguments()方法,其返回值类型是Map类型,当数据库调用这个getter方法时,数据库获取到的是一个Map类型的arguments,由于arguments在数据库中的存储类型是字符串,因此此时数据类型不匹配,获取到的Map类型的arguments无法存入数据库,就会出错。

所以我们需要在交换机类、队列类中里再提供一个返回值类型是字符串(String)的getArguments()方法,在该getter方法内部,将对象中是Map类型的arguments转化成字符串类型的arguments,此时数据库进行写数据时,调用getter方法将获取到的argumrnts是字符串类型的了,此时就可以将此字符串类型的argumrnts数据写入数据库中了,不会出错。

在这里插入图片描述

如果数据库想要进行读操作,就需要通过arguments变量的setter方法将从数据库中获取到的字符串类型的arguments设置给对象中的是Map类型的arguments,此时由于从数据库中获取到的arguments值是字符串类型的,而对象中的arguments是Map类型的,因此此时数据类型不匹配,无法设置,会出错。

所以我们需要在 交换机类、队列类中提供再提供一个参数是字符串类型的arguments,该setter方法内部使用JSON将字符串类型的arguments转化成一个Map类型的arguments,当数据库进行读操作时,就会调用这个setArguments(String arguments) 方法,就从数据库中获取到的字符串类型的arguments通过setter方法将其转换成Map类型的arguments,此时就不会出错了。

在这里插入图片描述

2.2 使用DataBaseManager类封装数据库操作

在我们实现 broker server 的代码中,进行数据库操作的类主要有 mqserver 下的mapper包下的MetaMapper接口,此时我们在mqserver包下新创建一个datacenter的包,datacenter包下创建一个DataBaseManager类,在该类中对数据库操作进行整合和调用。

该类中会定义一个初始化方法 init(),使用该方法进行数据库初始化:在 broker server 启动时进行逻辑判定,如果数据库已经存在,数据库表也已经存在,此时我们就不创建数据库了,只打印一个日志提醒数据库已经存在;如果数据库不存在,此时我们就创建数据库、创建表、插入一些默认的数据,同时打印日志提醒数据库初始化完成!

进行此逻辑判定是十分必要的,因为如果将 beoker server 部署到 一台新机器上,此时数据库是空的,所以需要进行建库、建表、插入默认数据等一系列操作;如果将 broker server 部署到 一台旧机器上,由于原先已经存在数据库,此时不再需要进行建库、建表、插入默认数据操作。

由于DataBaseManager类是进行数据库的封装,MataMapper中含有一系列数据库操作的方法,因此DataBaseManager类中需要引入MetaMapper实例。

DataBaseManager类中除了 metaMapper实例,init()初始化数据库方法外,还需要封装MetaMapper接口中的一系列操作数据库的方法,同时还有一个删除数据库的方法。

在这里插入图片描述
封装数据库操作这里还有一个需要注意的点:
**有没有发现引入DataaBaseManager类中的metaMapper实例是空引用?**我们从Spring框架中拿到Bean,一般都是使用注解@Autowired,但是此时我们并没有使用注解将MetaMapper的Bean拿到,所以此时写在DataBaseManager类中的metaMapper是个空对象。

那么有同学可能觉得,那就加个注解不就行了。但是只在MetaMapper对象上加还不行,外面使用它的类DataBaseManager也需要加上五大注解注册到Spring中,此时metaMapper对象才有效。

但是我们这里由于业务逻辑,打算自己管理这个类,因此并不打算将DataBaseManager这个类交给Spring管理。因此此时我们需要手动将MetaMapper类的Bean构造出来:
1、找到项目的入口启动类,在里面添加一个静态成员变量。
在这里插入图片描述
入口启动类:打开项目的src目录,点击main目录之后出现的XXApplication(是一个蓝色的带着C的图标),每个项目由于项目命名不同,其入口启动类都有差别,以自己的为准。

2、点击入口启动类,在类里面添加一个静态的成员变量context,同时在main方法中使用静态成员context接收run方法的返回值:
在这里插入图片描述

3、回到DataBaseManager类里的数据库初始化方法init(),对metaMapper对象进行手动获取:

metaMapper = MqApplication.context.getBean(MetaMapper.class);

此时metaMapper的Bean就获取到了,不再是空对象。

2.3 文件存储消息

由于咱们的消息数量庞大,且不需要频繁使用数据库进行增删改查,因此此时我们使用文件存储消息。一个文件是需要存储多条消息的,不是一个文件只存储一条消息哈!

由于一个文件可以存储多条消息,因此当我们想要从一个文件中获取某条消息时,就需要确定某条消息的位置,这会比较困难。因此我们在消息类(Message)中,定义了两个成员变量,分别是 offsetBeg、offsetEnd。offsetBeg 表示此条消息的头部到文件头部的距离,offsetEnd 表示此条消息的尾部到文件头部的距离。

在这里插入图片描述

2.3.1 存储消息时,按队列维度展开

那我们如何将消息存入文件呢??前面我们说过,生产者发送消息到 broker server 服务器之后,交换机(不同类型的交换机具有不同的转发规则)会将消息投递至当前与自己绑定的队列中。因此我们首先需要明确了,消息是依附于队列的,消息存储时,按照队列的维度展开。我们之前是将数据库文件meta.db存在目录data下的,目录data路径是在当前项目的相对路径下。

此时我们就在data目录下,创建出若干个子目录,每个子目录对应一个队列,子目录的名字以队列名命名进行区分。再在每个子目录下面创建两个文件,分别是 queue_data.txt(消息数据文件)该文件用于存储消息的内容、queue_stat.txt(消息统计文件),该文件用于存储消息的统计信息。

我们这样 约定:(自定义应用层)
1、queue_data.txt文件:该文件是一个二进制格式的文件,这个文件包含若干个消息,每个消息以二进制方式存储,每个消息由以下这几部分组成:
在这里插入图片描述
2、queue_stat.txt文件里,存储的是消息数据文件中消息的总数目以及消息的有效数目,此文件存储一行数据,但这一行数据有两列,譬如这样的:8000\t6800(消息的总内容数目/t消息的内容有效数目),意思是queue_data.txt文件中,消息的总数是8000,有效数目是6800,这两列数目使用 /t 分隔。

Message 对象在内存、硬盘中各存一份,内存中的那一份,要记录offsetBeg、offsetEnd,以便找到内存中的Message对象,找到内存中的Message对象,就能找到硬盘上的Message对象。

将数据存储一份在硬盘上,其实就是为了 当 broker server 重启时,内存上的数据消失不见后,能够将硬盘上持久化存储的数据读回到内存中,将原来的数据拿到。因此内存中的Message对象记录offsetBeg、offsetEnd后,就能方便的找到硬盘上的Message对象了。

2.3.1.1 逻辑删除

在这里我们探讨一下Message对象里isValid属性的作用。isValid 属性是用来判断 某条Message 是否 有效。对于 broker server 来说,消息既有新增也有删除。生产者生产一个消息,broker server 就需要新增一个消息;消费者消费掉一个消息,broker server 就需要删除一个消息。新增、删除在内存中比较容易(直接使用集合类操作即可),但是在文件上 增删比较困难。文件类似于一个 “顺序表”,插入比较简单,尾插即可,但是删除元素的话,就需要进行元素搬运,这个复杂程度不亚于O(N),甚至由于文件里消息数量庞大,复杂度比O(N)还高。因此我们使用 “逻辑删除” 去删除文件上的无效消息。

“逻辑删除” :逻辑删除不是真正的删除,只是将文件上的无效消息标记成无效

2.3.1.2 GC(垃圾回收)

使用逻辑删除衍生出来一个问题:要知道,逻辑删除只是一种标记手段,不是真正的删除,此时就会造成文件越来越大,但是实际里面的有效内容很少。此时,我们就要考虑进行GC(垃圾回收)。我们知道JVM为内存提供了GC,但是硬盘也可以GC。不过由于访问硬盘比访问内存低效,进行GC也就会比较低效。不过我们可以自己实现一个对硬盘的GC。

2.3.1.2.1 使用复制算法实现GC

此处的GC采用复制算法。啥是复制算法。

在这里插入图片描述

但在文件中,我们直接遍历消息数据文件,将消息数据文件里的有效消息拷贝出来,放到新文件queue_data_new.txt文件中,然后将原有的消息数据文件 queue_data.txt 删除,再将新文件 queue_data_new.txt文件重命名 成 queue_data.txt文件中。

确定了GC使用的策略后,啥时候触发一次GC呢?我们约定(此处是我的约定,你也可以约定别的情况下进行触发GC),当消息总数到达3000并且有效消息数小于60%时,触发GC。

2.3.1.3 文件存储消息的扩展内容(该功能暂未实现)

如果某个队列中,消息特别多,且有效消息占比很大,此时就会导致整个消息数据文件特别大。后续我们针对这个文件进行各种操作,成本就会上升很多。加入该文件大小是5G,此时触发一次GC其耗时很长,成本较高,影响系统性能。这时候我们考虑对有效数据多的大文件进行拆分,对无效数据多的消息进行GC,GC后的一个个小文件彼此相邻又可以进行合并。

2.3.1.3.1 文件拆分

当单个文件达到一定阈值之后,就会拆分成两个文件,此时拆着拆着就会成很多文件了。

2.3.1.3.2 文件合并

每个单独文件进行GC,GC过后文件变小了,就可以与相邻其他文件进行合并,此时可以防止大量小文件出现在硬盘上,浪费硬盘空间大小。

2.4 使用类MessageFileManager封装文件存储操作

三、将broker server 里的数据存储在内存上

这里先不讲。

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

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

相关文章

Visual Studio 2022进行文件差异比较

前言 Visual Studio 2022在版本17.7.4中发布在解决方案资源管理器中比较文件的功能&#xff0c;通过使用此功能&#xff0c;可以轻松地查看两个文件之间的差异&#xff0c;包括添加、删除和修改的代码行。可以逐行查看差异&#xff0c;并根据需要手动调整和编辑文件内容以进行…

100个openharmony开源demo:1.日历

准备用开发者手机写100个开源的demo不知道能不能实现&#xff0c;日拱一卒&#xff0c;期待蜕变。 第一个demo&#xff1a;日历&#xff0c;借鉴了网上的日历算法&#xff0c;自己用arkts写了界面和点击事件&#xff0c;各位可根据此demo写自己的日历选择器等组件。 1.目录结…

【Godot 3.5组件】简单血条组件HealthBar

说明 本文原文写自2022年&#xff0c;内容基于Godot3.5。是本人早期进行Godot组件化和自定义节点探索时的产物&#xff0c;当时的代码和思想可能不太成熟&#xff0c;但贴出来&#xff0c;供需要学习组件化基础思路的同学食用。 概述 血条作为一个非常基础和常见的组件&…

【C++】---string的模拟

【C】---string的模拟 一、string类实现1.string类的构造函数2.swap&#xff08;&#xff09;函数3.拷贝构造函数4.赋值运算符重载5.析构6.迭代器7.operator[ ]8.size9.c_str&#xff08;&#xff09;10.reserve&#xff08;&#xff09;11.resize&#xff08;&#xff09;12.p…

【C语言】linux内核pci_enable_device函数和_PCI_NOP宏

pci_enable_device 一、注释 static int pci_enable_device_flags(struct pci_dev *dev, unsigned long flags) {struct pci_dev *bridge;int err;int i, bars 0;/** 此时电源状态可能是未知的&#xff0c;可能是由于新启动或者设备移除调用。* 因此获取当前的电源状态&…

【Apache ShenYu源码】如何实现负载均衡模块设计

ShenYu是一个异步的&#xff0c;高性能的&#xff0c;跨语言的&#xff0c;响应式的 API 网关。有关ShenYu的介绍可以戳这。 一、前瞻 今天我们尝试不同的代码阅读方式&#xff0c;按模块来去阅读源码&#xff0c;看看效果如何。 本次阅读锁定在shenyu-loadbalancer&#xf…

WM8978 —— 带扬声器驱动程序的立体声编解码器(4)

接前一篇文章&#xff1a;WM8978 —— 带扬声器驱动程序的立体声编解码器&#xff08;3&#xff09; 九、寄存器概览与详解 1. 整体概览 WM8978芯片共有58个寄存器&#xff0c;整体总表如下&#xff1a; 2. 详细说明 在此&#xff0c;只介绍WM8978较为常用的那些寄存器。 &…

HBCalculator 程序:通过 VMD 可计算分子动力学模拟中氢键密度和强度的一维和二维分布

分享一个通过 VMD 可计算分子动力学模拟中氢键密度和强度的一维和二维分布程序 HBCalculator。 感谢论文的原作者&#xff01; 主要内容 “氢键是分子系统中关键的非共价相互作用&#xff0c;对生物、化学和能量相关过程产生重大影响&#xff1b;因此&#xff0c;描述氢键信息…

Eureka的介绍和作用,以及搭建

一、Eureka的介绍和作用 Eureka是Netflix开源的一种服务发现和注册工具&#xff0c;它为分布式系统中的服务提供了可靠的服务发现和故障转移能力。Eureka是Netflix的微服务架构的关键组件之一&#xff0c;它能够实时地监测和管理服务实例的状态和可用性。 在Eureka架构中&…

智能合约 之 ERC-20介绍

什么是ERC20 ERC20全称为Ethereum Request for Comment 20&#xff0c;是一种智能合约标准&#xff0c;用于以太坊网络上的代币发行 姊妹篇 - 如何部署ERC20 ERC20的应用场景 代币化资产&#xff0c;例如&#xff1a;USDT 是一种以美元为背书的ERC20代币&#xff0c;每个USDT代…

scDEA一键汇总12种单细胞差异分析方法 DESeq2、edgeR、MAST、monocle、scDD、Wilcoxon

问题来源 单细胞可以做差异分析&#xff0c;但是究竟选择哪种差异分析方法最靠谱呢&#xff1f; 解决办法 于是我去检索文献&#xff0c;是否有相关研究呢&#xff1f; https://academic.oup.com/bib/article/23/1/bbab402/6375516 文章指出&#xff0c;现有的差异分析方法…

Java推荐算法——特征加权推荐算法(以申请学校为例)

加权推荐算法 文章目录 加权推荐算法1.推荐算法的简单介绍2.加权推荐算法详细介绍3.代码实现4.总结 1.推荐算法的简单介绍 众所周知&#xff0c;推荐算法有很多种&#xff0c;例如&#xff1a; 1.加权推荐&#xff1a;分为简单的特征加权&#xff0c;以及复杂的混合加权。主要…

23 OpenCV 直方图比较

文章目录 直方图比较的目的相关性计算 (CV_COMP_CORREL)卡方计算 (CV_COMP_CHISQR)十字计算(CV_COMP_INTERSECT)巴氏距离计算 (CV_COMP_BHATTACHARYYA )compareHist 直方图比较算子示例 直方图比较的目的 直方图比较的目的是衡量两幅图像之间的相似度或差异度。通过计算图像的颜…

Vue 3中实现基于角色的权限认证实现思路

一、基于角色的权限认证主要步骤 在Vue 3中实现基于角色的权限认证通常涉及以下几个主要步骤&#xff1a; 定义角色和权限&#xff1a;首先需要在后端服务定义不同的角色和它们对应的权限。权限可以是对特定资源的访问权限&#xff0c;比如读取、写入、修改等。用户认证&#…

C#非强签名dll搜索顺序

由于不是强签名dll&#xff0c;所以无效考虑全局程序集缓存 (GAC)。 预备工作 新建解决方案ClassLibrary1,新建类库ClassLibrary1,新建控制台程序ShowDllLoc。 利用VS添加引用。 一&#xff0c;利用app.config设置codebase&#xff0c;设置dll的加载路径为&#xff1a;code…

前后端分离项目springsecurity实现用户登录认证快速使用

目录 1、引入依赖 2、创建类继承WebSecurityConfigurerAdapter &#xff08;1&#xff09;重写里面的configure(HttpSecurity http)方法 &#xff08;2&#xff09;重写AuthenticationManager authenticationManagerBean() &#xff08;3&#xff09;密码加密工具 3、继承…

市场复盘总结 20240321

仅用于记录当天的市场情况&#xff0c;用于统计交易策略的适用情况&#xff0c;以便程序回测 短线核心&#xff1a;不参与任何级别的调整&#xff0c;采用龙空龙模式 一支股票 10%的时候可以操作&#xff0c; 90%的时间适合空仓等待 二进三&#xff1a; 进级率中 23% 最常用的…

算法体系-14 第十四 贪心算法(上)

一 、 递归套路解决判断完全二叉树 1.1 描述 1.2 分析 1.3 代码 public static boolean isCBT2(Node head) {return process(head).isCBT;}public static class Info {public boolean isFull;public boolean isCBT;public int height;public Info(boolean full, boolean cbt…

JNDI注入原理及利用IDEA漏洞复现

&#x1f36c; 博主介绍&#x1f468;‍&#x1f393; 博主介绍&#xff1a;大家好&#xff0c;我是 hacker-routing &#xff0c;很高兴认识大家~ ✨主攻领域&#xff1a;【渗透领域】【应急响应】 【Java、PHP】 【VulnHub靶场复现】【面试分析】 &#x1f389;点赞➕评论➕收…

【题目】【网络系统管理】2019年全国职业技能大赛高职组计算机网络应用赛项H卷

极安云科专注职业教育技能竞赛培训4年&#xff0c;包含信息安全管理与评估、网络系统管理、网络搭建等多个赛项及各大CTF模块培训学习服务。本团队基于赛项知识点&#xff0c;提供完整全面的系统性理论教学与技能培训&#xff0c;成立至今持续优化教学资源与讲师结构&#xff0…