diff --git a/CHANGELOG.md b/CHANGELOG.md index 9cf55af..4d8a23b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,18 @@ ``` ------ +## [V 1.0.2] - 2023.4.24 +### common +- 🎈新增: 新增 `ConfigFile` 方法 `onlyUpdateTime,updateConfigTime` 负责更新外部数据上传时间和配置文件本身的上传时间 +- 🎈新增: 新增 `TimeUtil` 工具类，用于获取LocalDateTime的秒数 +- 🎈新增: 新增 `FileCacheException` 异常类，用于处理文件池异常 + +### FileModule +- 🎈新增: 新增 `FileCache` 文件缓冲池类，负责缓存文件内容，文件的读取，修改，追加，能够根据刷入时间或者写入字节，来进行自动刷盘操作 +- 🎈新增: 新增 `FileCacheManager` 文件缓冲池管理类，管理所有文件缓存池，轮询查看每个文件是否需要自动刷入，目前包含巡逻线程与刷入线程 + +------ + ## [V 1.0.1] - 2023.4.21 ### common - 🎈新增: 新增 `ConstPool` 常量池，用于存放常量，目前存放了模块名称常量，便于开发统一 diff --git a/FileModule/src/main/java/org/example/bean/ModuleSrcConfigFile.java b/FileModule/src/main/java/org/example/bean/ModuleSrcConfigFile.java index 0dec81b..35cac06 100644 --- a/FileModule/src/main/java/org/example/bean/ModuleSrcConfigFile.java +++ b/FileModule/src/main/java/org/example/bean/ModuleSrcConfigFile.java @@ -13,7 +13,7 @@ import java.util.Map; public class ModuleSrcConfigFile extends ConfigFile> { - private static final Map config; + private static Map config; public static class SRC{ private String src; @@ -24,6 +24,8 @@ public class ModuleSrcConfigFile extends ConfigFile packageConfig() { return super.packageConfig(); } + } diff --git a/FileModule/src/main/java/org/example/cache/FileCache.java b/FileModule/src/main/java/org/example/cache/FileCache.java new file mode 100644 index 0000000..9ca784c --- /dev/null +++ b/FileModule/src/main/java/org/example/cache/FileCache.java @@ -0,0 +1,232 @@ +package org.example.cache; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import org.example.common.ConfigFile; +import org.example.exception.FileCacheException; +import org.example.util.JsonFileUtil; +import org.example.util.TimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.nio.file.Paths; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * @author Genius + * @date 2023/04/24 00:01 + **/ + +//TODO 文件缓存写入优化 考虑是否要加一个缓存写入Buffer，将短时间内多个相同Key的内容存入Buffer中，合并存入 +public class FileCache { + private T configFile; //文件配置类，得到文件信息和文件夹结构，更新文件存入时间 + private Logger logger; + + private ConcurrentHashMap jsonFile; //文件内容缓存 + + private ConcurrentHashMap oldJsonFileTemp; //保存上一个版本的文件内容，用于优化自动刷入 + private static final int MAX_WRITE_BUFFER_LIMIT = 4096; //最大写入缓存上线 + + private AtomicInteger writeByte; //当前写入的字节数 + + private BlockingQueue> syncChannel; //磁盘刷入阻塞队列 + + //TODO 优化 考虑是否采用一个定时线程管理所有FileCache的写入 + private ExecutorService pool; //Sync线程池 + + private long autoSyncTime; //自动刷入时间 + + public FileCache(T configFile) throws FileCacheException { + init(configFile,10); + } + + public FileCache(T configFile,long autoSyncTime)throws FileCacheException { + init(configFile,autoSyncTime); + } + + /** + * 初始化方法 + * @param configFile 配置文件 + * @param autoSyncTime 自动刷入时间 + * @throws FileCacheException + */ + private void init(T configFile, long autoSyncTime) throws FileCacheException { + this.configFile = configFile; + this.configFile.updateConfigTime();//更新一下当前的时间 + + this.logger = LoggerFactory.getLogger("FileCache:"+this.configFile.getFileName()); + + this.autoSyncTime = autoSyncTime; + if(!load(getFileName())){ + throw new FileCacheException("FileCache Init Error,please Check if your path is correct"); + } + + this.writeByte = new AtomicInteger(0); + this.syncChannel = new ArrayBlockingQueue<>(20); + this.pool = Executors.newSingleThreadExecutor(); + pool.submit(new SyncMan()); + } + + /** + * 加载文件内容 + * @return boolean + */ + private boolean load(String path){ + Map stringObjectMap = JsonFileUtil.readJsonFile(path); + if(Objects.isNull(stringObjectMap)){ + logger.error("{}配置文件不存在!",path); + return false; + } + + this.jsonFile = new ConcurrentHashMap<>(stringObjectMap); + this.oldJsonFileTemp = new ConcurrentHashMap<>(stringObjectMap); + return true; + } + + /** + * 重新加载内存池Map + * @return boolean + */ + public synchronized boolean reload(){ + return load(Paths.get(this.configFile.getFilePath(),this.configFile.getFileName()).toString()); + } + + //写入 + public int write(String key,Object data) throws InterruptedException { + String jsonDataStr = JSON.toJSONString(data); + int writeBytes = key.getBytes().length + jsonDataStr.getBytes().length; + + JSONObject jsonData = writeInData(key, data); + jsonFile.put("data",jsonData); + + ConcurrentHashMap temp = new ConcurrentHashMap<>(jsonFile); + int newBytes = writeByte.updateAndGet(x -> x + writeBytes >= MAX_WRITE_BUFFER_LIMIT ? 0 : x + writeBytes); + + //TODO 此处会发生脏读问题，即put进入的Map版本不是当前版本，但是目前没有发现该问题是否会影响到文件写入 + if(newBytes==0){ + logger.debug("缓冲区已满，刷入磁盘"); + syncChannel.put(temp); + } + return writeBytes; + } + + /** + * 追加内容 + * @param key key + * @param append 追加内容 + * @return + * @throws InterruptedException + */ + public int append(String key,Object append) throws InterruptedException { + Object data = getData(key); + StringBuffer buffer = new StringBuffer(JSON.toJSONString(data)); + String jsonStr = buffer.append(JSON.toJSONString(append)).toString(); + return write(key, jsonStr); + } + + private JSONObject writeInData(String key,Object value){ + JSONObject data = JSONObject.parseObject(jsonFile.get("data").toString()); //此处必须返回一个新的JsonObject，否则会导致旧版本同步更新 + data.put(key,value); + return data; + } + + /** + * 获取文件内容 + * @param key + * @return + */ + public Object get(String key){ + return jsonFile.get(key); + } + + /** + * 获取文件数据内容 + * @param key + * @return + */ + public Object getData(String key){ + Object data = this.get("data"); + JSONObject jsonObject = JSONObject.parseObject(data.toString()); + return jsonObject.get(key); + } + + /** + * 清除已写入的字节数记录 + */ + protected void clearWriteBytes(){ + writeByte.updateAndGet(x->0); + } + + /** + * 判断当前时间是否超过更新时间 + * @return boolean + */ + public boolean needAutoSync(){ + long now = TimeUtil.getCurrentSecond(); + return now- TimeUtil.getSecond(configFile.getUpdateTime())>autoSyncTime; + } + + /** + * 强制刷入磁盘 + */ + public void forceSync(){ + clearWriteBytes(); + ConcurrentHashMap temp = new ConcurrentHashMap<>(jsonFile); + try { + syncChannel.put(temp); + } catch (InterruptedException e) { + logger.debug("自动刷入失败"); + } + } + + /** + * 超过缓冲区刷入 + * @return + */ + private boolean sync(ConcurrentHashMap take){ + configFile.updateConfigTime(); //刷新配置文件刷入时间 + String dir = getFileName(); + if(oldJsonFileTemp.get("data").toString().equals(take.get("data").toString())){ + logger.debug("未发生版本变化"); + return true; + } + + configFile.onlyUpdateTime(take); + oldJsonFileTemp = new ConcurrentHashMap<>(take); + File file = JsonFileUtil.writeJsonFile(dir, take); + logger.debug("正在写入{}新版本",dir); + return Objects.isNull(file); + } + + public BlockingQueue getFileChannel(){ + return this.syncChannel; + } + + public String getFileName(){ + return Paths.get(this.configFile.getFilePath(), this.configFile.getFileName()).toString(); + } + + public long getSyncTime(){ + return this.autoSyncTime; + } + + class SyncMan implements Runnable{ + + @Override + public void run() { + for(;;){ + try { + ConcurrentHashMap take = syncChannel.take(); + sync(take); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + } + } + +} diff --git a/FileModule/src/main/java/org/example/cache/FileCacheManager.java b/FileModule/src/main/java/org/example/cache/FileCacheManager.java new file mode 100644 index 0000000..641205a --- /dev/null +++ b/FileModule/src/main/java/org/example/cache/FileCacheManager.java @@ -0,0 +1,103 @@ +package org.example.cache; + +import org.example.util.TimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import static java.lang.Thread.sleep; + +/** + * @author Genius + * @date 2023/04/24 17:35 + **/ + +/** + * 文件自动刷入管理类，不断监听文件是否需要自动写入 + */ +public class FileCacheManager { + + private Logger logger = LoggerFactory.getLogger(FileCacheManager.class); + private final List fileCaches; + + private long sleepTime; //睡眠时间 + + private ExecutorService watchPool; //巡逻线程 + + private ExecutorService autoSyncer; //生产者线程 + + public FileCacheManager(List fileCaches){ + this.fileCaches = fileCaches; + initSleepTime(); + this.watchPool = Executors.newSingleThreadExecutor(); + this.autoSyncer = Executors.newFixedThreadPool(fileCaches.size()); + } + + /** + * 根据文件缓存的刷盘时间得到一个最小的睡眠时间，减少空转 + */ + private void initSleepTime(){ + AtomicLong minSleepTime = new AtomicLong(Long.MAX_VALUE); + fileCaches.forEach(item->{ + minSleepTime.set(Long.min(minSleepTime.get(), item.getSyncTime())); + }); + this.sleepTime = minSleepTime.get(); + } + + public void start(){ + this.watchPool.submit(new Watcher()); + } + + class Watcher implements Runnable{ + + @Override + public void run() { + for(;;){ + long now = TimeUtil.getCurrentSecond(); + for(FileCache cache:fileCaches){ + BlockingQueue fileChannel = cache.getFileChannel(); + if(fileChannel.isEmpty()){ + if(cache.needAutoSync()){ + logger.debug("检测到需要强制刷新的文件 {}",cache.getFileName()); + autoSyncer.submit(new AutoSyncer(cache)); + } + } + } + now -= TimeUtil.getCurrentSecond(); + if(now + + com.alibaba + fastjson + 1.2.75 + junit junit 3.8.1 test + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/common/src/main/java/org/example/common/ConfigFile.java b/common/src/main/java/org/example/common/ConfigFile.java index 78fa6c4..17538a0 100644 --- a/common/src/main/java/org/example/common/ConfigFile.java +++ b/common/src/main/java/org/example/common/ConfigFile.java @@ -9,26 +9,58 @@ import java.util.Map; * @date 2023/04/21 02:24 **/ -//配置文件的抽象类 +//配置文件的抽象类，只负责构建配置文件最基础的架构，一般不用来存放配置文件本身的内容 public abstract class ConfigFile { private String filePath; private String fileName; - private T data; + private T data; //json文件的结构不是文件的数据 例如 {username:"",password:""} + + //上一次更新时间 + private LocalDateTime updateTime; /** - * 用于打包配置文件 + * 用于最开始创建配置文件结构的打包 + * @return Map */ public Map packageConfig() { + return this.packageConfig(this.data); + } - + /** + * 用于给外部函数提供的内容打包 + * @return + */ + public Map packageConfig(T data){ + updateConfigTime(); return Map.of( "data",data, - "updateTime", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "updateTime", updateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) ); } + /** + * 只进行时间更新操作 + * @param map + * @return + */ + public Map onlyUpdateTime(Map map){ + updateConfigTime(); + if (map.containsKey("updateTime")) { + map.put("updateTime",updateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); + } + return map; + } + + /** + * 更新配置文件类本身的时间 + */ + public void updateConfigTime(){ + updateTime = LocalDateTime.now(); + } + + public ConfigFile() { } @@ -36,6 +68,7 @@ public abstract class ConfigFile { this.filePath = filePath; this.fileName = fileName; this.data = data; + this.updateTime = LocalDateTime.now(); } public String getFilePath() { @@ -46,7 +79,12 @@ public abstract class ConfigFile { return fileName; } - protected T getData() { + public LocalDateTime getUpdateTime() {return updateTime;} + + //不推荐直接使用 + public T getData() { return data; } + + public void setData(T data){this.data = data;} } diff --git a/common/src/main/java/org/example/exception/FileCacheException.java b/common/src/main/java/org/example/exception/FileCacheException.java new file mode 100644 index 0000000..f8c4374 --- /dev/null +++ b/common/src/main/java/org/example/exception/FileCacheException.java @@ -0,0 +1,19 @@ +package org.example.exception; + +/** + * @author Genius + * @date 2023/04/24 00:57 + **/ +public class FileCacheException extends Exception{ + + String message; + + public FileCacheException(String ErrorMessage){ + this.message = ErrorMessage; + } + + @Override + public String getMessage() { + return message; + } +} diff --git a/common/src/main/java/org/example/util/TimeUtil.java b/common/src/main/java/org/example/util/TimeUtil.java new file mode 100644 index 0000000..93ccac7 --- /dev/null +++ b/common/src/main/java/org/example/util/TimeUtil.java @@ -0,0 +1,19 @@ +package org.example.util; + +import java.time.LocalDateTime; +import java.time.ZoneOffset; + +/** + * @author Genius + * @date 2023/04/24 18:47 + **/ +public class TimeUtil { + + public static long getCurrentSecond(){ + return LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8")); + } + + public static long getSecond(LocalDateTime dateTime){ + return dateTime.toEpochSecond(ZoneOffset.of("+8")); + } +} diff --git a/console/pom.xml b/console/pom.xml index cb4a981..5894b60 100644 --- a/console/pom.xml +++ b/console/pom.xml @@ -33,6 +33,11 @@ FileModule 1.0-SNAPSHOT + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/console/src/main/java/org/example/ConsoleApplication.java b/console/src/main/java/org/example/ConsoleApplication.java index 996e45d..b585518 100644 --- a/console/src/main/java/org/example/ConsoleApplication.java +++ b/console/src/main/java/org/example/ConsoleApplication.java @@ -1,6 +1,9 @@ package org.example; +import org.example.bean.ModuleSrcConfigFile; +import org.example.cache.FileCache; +import org.example.exception.FileCacheException; import org.example.init.InitWorld; import org.example.init.ModuleSrcConfigFileInit; import org.springframework.boot.SpringApplication; @@ -16,11 +19,12 @@ import java.util.List; @SpringBootApplication public class ConsoleApplication { - public static void main(String[] args) { + public static void main(String[] args) throws FileCacheException { if (InitWorld.getInstance() .setInitMachines(List.of(new ModuleSrcConfigFileInit())) .start()) { + SpringApplication.run(ConsoleApplication.class, args); }