net.sz.framework 框架 轻松搭建数据服务中心----读写分离数据一致性,滑动缓存
前言
前文讲述了net.sz.framework 框架的基础实现功能,本文主讲 net.sz.framework.db 和 net.sz.framework.szthread;
net.sz.framework.db 是 net.sz.framework 底层框架下的orm框架,仿照翻译了hibernate实现功能,虽然不足hibernate强大;但在于其功能实现单一高效和高可控性;
net.sz.framework.szthread 是 net.sz.framework 底层框架下的线程控制中心和线程池概念;
以上就不在赘述,前面的文章已经将结果了;
叙述
无论你是做何种软件开发,都离不开数据;
数据一般我们都会有两个问题一直在脑后徘徊,那就是读和写的问题;
一般正常情况下数据我们可能出现的存储源是数据库(mysql,sqlserver,sqlite,Nosql等)、文件数据库(excel,xml,cvs等)
无论是合作数据格式都只是在意数据的存储;保证数据不丢失等情况;
那么我们为了数据的读取和写入高效会想尽办法去处理数据,已达到我们需求范围类的数据最高效最稳当的方式;
今天我们准备的是 orm框架下面的 SqliteDaoImpl 对 sqlite数据源 进行测试和代码设计;换其他数据源也是大同小异;
准备工作
新建项目 maven java项目 net.sz.dbserver
我们在项目下面创建model、cache、db、main这几个包;
然后在 model 包 下面创建 ModelTest 类

然后在db包下面建立dbmanager类;
1 package net.sz.dbserver.db; 2 3 import java.sql.Connection; 4 import java.util.ArrayList; 5 import net.sz.dbserver.model.ModelTest; 6 import net.sz.framework.db.Dao; 7 import net.sz.framework.db.SqliteDaoImpl; 8 import net.sz.framework.szlog.SzLogger; 9 import net.sz.framework.utils.PackageUtil; 10 11 /** 12 * 13 * <br> 14 * author 失足程序员<br> 15 * blog http://www.cnblogs.com/ty408/<br> 16 * mail 492794628@qq.com<br> 17 * phone 13882122019<br> 18 */ 19 public class DBManager { 20 21 private static SzLogger log = SzLogger.getLogger(); 22 private static final DBManager IN_ME = new DBManager(); 23 24 public static DBManager getInstance() { 25 return IN_ME; 26 } 27 28 Dao dao = null; 29 30 public DBManager() { 31 try { 32 /*不使用连接池,显示执行sql语句的数据库操作*/ 33 this.dao = new SqliteDaoImpl("/home/sqlitedata/testdb.dat", true); 34 } catch (Exception e) { 35 log.error("创建数据库连接", e); 36 } 37 } 38 39 /** 40 * 检查并创建数据表结构 41 */ 42 public void checkTables() { 43 /*创建连接,并自动释放*/ 44 try (Connection con = this.dao.getConnection()) { 45 String packageName = "net.sz.dbserver.model"; 46 /*获取包下面所有类*/ 47 ArrayList<Class<?>> tables = PackageUtil.getClazzs(packageName); 48 if (tables != null) { 49 for (Class<?> table : tables) { 50 /*检查是否是需要创建的表*/ 51 if (this.dao.checkClazz(table)) { 52 /*创建表结构*/ 53 this.dao.createTable(con, table); 54 } 55 } 56 } 57 } catch (Exception e) { 58 log.error("创建表抛异常", e); 59 } 60 } 61 62 }
我们在dbmanager类里面通过SqliteDaoImpl 类创建了sqlite数据库支持的类似于hibernate的辅助;
在checktables下面会查找我们项目包下面所有类型,并且创建数据表;如果表存在就更新表结构(sqlite特性,不会更新表结构);
我们在checktables函数下面做到了对连接的复用情况;创建后并自动释放代码
接下来main包里面创建主函数启动类

以上代码我们完成了数据库文件和数据表的创建
1 --- exec-maven-plugin:1.2.1:exec (default-cli) @ net.sz.dbserver --- 2 设置系统字符集sun.stdout.encoding:utf-8 3 设置系统字符集sun.stderr.encoding:utf-8 4 日志级别:DEBUG 5 输出文件日志目录:../log/sz.log 6 是否输出控制台日志:true 7 是否输出文件日志:true 8 是否使用双缓冲输出文件日志:true 9 [04-07 10:56:38:198:ERROR:MainManager.main():19] 创建数据库,创建数据表结构 10 [04-07 10:56:38:521:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.model.ModelTest 字段:log is transient or static or final; 11 [04-07 10:56:38:538:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 无此表 12 [04-07 10:56:38:561:ERROR:SqliteDaoImpl.createTable():200] 13 表: 14 create table if not exists `ModelTest` ( 15 `Id` bigint not null primary key, 16 `name` varchar(255) null 17 ); 18 创建完成;
这里的步骤在之前文章《存在即合理,重复轮子orm java版本》里面有详细介绍,不过当前版本和当时文章版本又有更多优化和改进;
准备测试数据
1 /*创建支持id*/ 2 GlobalUtil.setServerID(1); 3 for (int i = 0; i < 10; i++) { 4 ModelTest modelTest = new ModelTest(); 5 /*获取全局唯一id*/ 6 modelTest.setId(GlobalUtil.getId()); 7 /*设置参数*/ 8 modelTest.setName("123"); 9 10 try { 11 DBManager.getInstance().getDao().insert(modelTest); 12 } catch (Exception e) { 13 log.error("写入数据失败", e); 14 } 15 }
输出

重构modeltest类
首先在cache包下面创建CacheBase类实现缓存的基本参数
1 package net.sz.dbserver.cache; 2 3 import javax.persistence.Id; 4 import net.sz.framework.util.AtomInteger; 5 6 /** 7 * 8 * <br> 9 * author 失足程序员<br> 10 * blog http://www.cnblogs.com/ty408/<br> 11 * mail 492794628@qq.com<br> 12 * phone 13882122019<br> 13 */ 14 public class CacheBase { 15 16 /*主键ID*/ 17 @Id 18 protected long Id; 19 20 /*编辑状态 是 transient 字段,不会更新到数据库的*/ 21 private volatile transient boolean edit; 22 /*版本号 是 transient 字段,不会更新到数据库的*/ 23 private volatile transient AtomInteger versionId; 24 /*创建时间*/ 25 private volatile transient long createTime; 26 /*最后获取缓存时间*/ 27 private volatile transient long lastGetCacheTime; 28 29 public CacheBase() { 30 } 31 32 /** 33 * 创建 34 */ 35 public void createCache() { 36 edit = false; 37 versionId = new AtomInteger(1); 38 createTime = System.currentTimeMillis(); 39 lastGetCacheTime = System.currentTimeMillis(); 40 } 41 42 public long getId() { 43 return Id; 44 } 45 46 public void setId(long Id) { 47 this.Id = Id; 48 } 49 50 public boolean isEdit() { 51 return edit; 52 } 53 54 public void setEdit(boolean edit) { 55 this.edit = edit; 56 } 57 58 public AtomInteger getVersionId() { 59 return versionId; 60 } 61 62 public void setVersionId(AtomInteger versionId) { 63 this.versionId = versionId; 64 } 65 66 public long getCreateTime() { 67 return createTime; 68 } 69 70 public void setCreateTime(long createTime) { 71 this.createTime = createTime; 72 } 73 74 public long getLastGetCacheTime() { 75 return lastGetCacheTime; 76 } 77 78 public void setLastGetCacheTime(long lastGetCacheTime) { 79 this.lastGetCacheTime = lastGetCacheTime; 80 } 81 82 /** 83 * 拷贝数据 84 * 85 * @param cacheBase 86 */ 87 public void copy(CacheBase cacheBase) { 88 this.Id = cacheBase.Id; 89 } 90 91 }
在cachebase类中,我创建了copy函数用来赋值新数据的;
通过这个类型,我们可以做到定时缓存,滑动缓存效果;
增加版号的作用在于,更新操作标识,是否是编辑状态也是用作更新标识;
于此同时我们把原 ModelTest 唯一键、主键 id 移动到了 cachebase 父类中
修改modeltest类继承cachebase;
1 public class ModelTest extends CacheBase
改造一下dbmanager
1 Dao readDao = null; 2 Dao writeDao = null; 3 4 public Dao getReadDao() { 5 return readDao; 6 } 7 8 public Dao getWriteDao() { 9 return writeDao; 10 } 11 12 public DBManager() { 13 try { 14 /*不使用连接池,显示执行sql语句的数据库操作*/ 15 this.readDao = new SqliteDaoImpl("/home/sqlitedata/testdb.dat", true); 16 /*不使用连接池,显示执行sql语句的数据库操作*/ 17 this.writeDao = new SqliteDaoImpl("/home/sqlitedata/testdb.dat", true); 18 } catch (Exception e) { 19 log.error("创建数据库连接", e); 20 } 21 }
加入读取数据库连接、写入数据库连接;
CacheManager
在cache包下面建立cachemanager类;
cachemanager 类型是我们具体和重点思路;
构建了读取,并加入缓存集合;
构建了更新并写入数据库;
同时读取和更新都保证线程安全性特点;
1 package net.sz.dbserver.cache; 2 3 import java.util.concurrent.ConcurrentHashMap; 4 import net.sz.dbserver.db.DBManager; 5 import net.sz.framework.szlog.SzLogger; 6 7 /** 8 * 9 * <br> 10 * author 失足程序员<br> 11 * blog http://www.cnblogs.com/ty408/<br> 12 * mail 492794628@qq.com<br> 13 * phone 13882122019<br> 14 */ 15 public class CacheManager { 16 17 private static SzLogger log = SzLogger.getLogger(); 18 private static final CacheManager IN_ME = new CacheManager(); 19 20 public static CacheManager getInstance() { 21 return IN_ME; 22 } 23 /*缓存集合*/ 24 final ConcurrentHashMap<Long, CacheBase> cacheMap = new ConcurrentHashMap<>(); 25 26 /** 27 * 获取一条数据,这里我只是测试,提供思路, 28 * <br> 29 * 所以不会去考虑list等情况; 30 * <br> 31 * 需要的话可以自行修改 32 * 33 * @param <T> 34 * @param clazz 35 * @param id 36 * @return 37 */ 38 public <T extends CacheBase> T getCacheBase(Class<T> clazz, long id) { 39 CacheBase cacheBase = null; 40 cacheBase = cacheMap.get(id); 41 if (cacheBase == null) { 42 try { 43 /*先读取数据库*/ 44 cacheBase = DBManager.getInstance().getReadDao().getObjectByWhere(clazz, "where id=@id", id); 45 /*加入同步操作*/ 46 synchronized (cacheMap) { 47 /*这个时候再次读取缓存,防止并发*/ 48 CacheBase tmp = cacheMap.get(id); 49 /*双重判断*/ 50 if (tmp == null) { 51 /*创建缓存标识*/ 52 cacheBase.createCache(); 53 /*加入缓存信息*/ 54 cacheMap.put(id, cacheBase); 55 } else { 56 cacheBase = tmp; 57 } 58 } 59 } catch (Exception e) { 60 log.error("读取数据异常", e); 61 } 62 } 63 64 if (cacheBase != null) { 65 /*更新最后获取缓存的时间*/ 66 cacheBase.setLastGetCacheTime(System.currentTimeMillis()); 67 } 68 69 return (T) cacheBase; 70 } 71 72 /** 73 * 更新缓存数据同时更新数据库数据 74 * 75 * @param <T> 76 * @param t 77 * @return 78 */ 79 public <T extends CacheBase> boolean updateCacheBase(T t) { 80 if (t == null) { 81 throw new UnsupportedOperationException("参数 T 为 null"); 82 } 83 try { 84 CacheBase cacheBase = null; 85 cacheBase = cacheMap.get(t.getId()); 86 /*理论上,控制得当这里是不可能为空的*/ 87 if (cacheBase != null) { 88 /*理论上是能绝对同步的,你也可以稍加修改*/ 89 synchronized (cacheBase) { 90 /*验证编辑状态和版号,保证写入数据是绝对正确的*/ 91 if (cacheBase.isEdit() 92 && cacheBase.getVersionId() == t.getVersionId()) { 93 /*拷贝最新数据操作*/ 94 cacheBase.copy(t); 95 /*写入数据库,用不写入还是同步写入,看自己需求而一定*/ 96 DBManager.getInstance().getWriteDao().update(cacheBase); 97 /*保证写入数据库后进行修改 对版本号进行加一操作*/ 98 cacheBase.getVersionId().changeZero(1); 99 /*设置最新的最后访问时间*/ 100 cacheBase.setLastGetCacheTime(System.currentTimeMillis()); 101 /*修改编辑状态*/ 102 cacheBase.setEdit(false); 103 log.error("数据已修改,最新版号:" + cacheBase.getVersionId()); 104 return true; 105 } else { 106 log.error("版本已经修改无法进行更新操作"); 107 throw new UnsupportedOperationException("版本已经修改无法进行更新操作"); 108 } 109 } 110 } else { 111 log.error("缓存不存在无法修改数据"); 112 throw new UnsupportedOperationException("缓存不存在无法修改数据"); 113 } 114 } catch (Exception e) { 115 throw new UnsupportedOperationException("更新数据异常", e); 116 } 117 } 118 119 /** 120 * 获取独占编辑状态 121 * 122 * @param id 123 * @return 124 */ 125 public boolean updateEdit(long id) { 126 CacheBase t = null; 127 t = cacheMap.get(id); 128 if (t == null) { 129 throw new UnsupportedOperationException("未找到数据源"); 130 } 131 return updateEdit(t); 132 } 133 134 /** 135 * 获取独占编辑状态 136 * 137 * @param t 138 * @return 139 */ 140 public boolean updateEdit(CacheBase t) { 141 if (t == null) { 142 throw new UnsupportedOperationException("参数 T 为 null"); 143 } 144 if (!t.isEdit()) { 145 synchronized (t) { 146 if (!t.isEdit()) { 147 /*同步后依然需要双重判定*/ 148 t.setEdit(true); 149 return true; 150 } 151 } 152 } 153 return false; 154 } 155 156 }
可能有人要问, 为啥要加锁,加版号或者加编辑状态;
我们先看一张图片
当同一份数据,展示给客户端(web,多线程等)的时候,同时进行获取,进行编辑,我们不可能每次都需要去调用独占编辑;
那么问题来了我们就拿modeltest的name字段说明,当前等于123,当client1和client2都表示数据的名字错误了需要修改成789;
那么在写入数据库的时候总会有先后顺序,那么后面的很可能就覆盖了前面的修改,
我们假如client1先提交,把name字段改为456,这时候client2提交了,789就直接覆盖了456字段,
程序根本不知道字段的覆盖了,也不知道哪一个是正确的;
所以我加入了编辑状态和版号验证;当然你也可以根据你的需求来进行修改
1 package net.sz.dbserver.main; 2 3 import net.sz.dbserver.cache.CacheManager; 4 import net.sz.dbserver.db.DBManager; 5 import net.sz.dbserver.model.ModelTest; 6 import net.sz.framework.szlog.SzLogger; 7 import net.sz.framework.utils.GlobalUtil; 8 9 /** 10 * 11 * <br> 12 * author 失足程序员<br> 13 * blog http://www.cnblogs.com/ty408/<br> 14 * mail 492794628@qq.com<br> 15 * phone 13882122019<br> 16 */ 17 public class MainManager { 18 19 private static SzLogger log = SzLogger.getLogger(); 20 21 public static void main(String[] args) { 22 23 log.error("创建数据库,创建数据表结构"); 24 DBManager.getInstance().checkTables(); 25 /*创建支持id*/ 26 GlobalUtil.setServerID(1); 27 ModelTest modelTest = new ModelTest(); 28 /*获取全局唯一id*/ 29 modelTest.setId(GlobalUtil.getId()); 30 /*设置参数*/ 31 modelTest.setName("123"); 32 33 /*创建测试数据先修改数据库*/ 34 try { 35 DBManager.getInstance().getReadDao().insert(modelTest); 36 } catch (Exception e) { 37 log.error("写入数据失败", e); 38 } 39 40 /*打印一次id*/ 41 log.error("modelTest.getId()=" + modelTest.getId()); 42 43 for (int i = 0; i < 3; i++) { 44 new Thread(() -> { 45 try { 46 /*上面的写入数据是为了获取这个id,保证测试代码编辑功能*/ 47 ModelTest cacheBase = CacheManager.getInstance().getCacheBase(ModelTest.class, modelTest.getId()); 48 if (cacheBase != null) { 49 log.error("成功获得数据"); 50 /*独占编辑状态你可以不需要*/ 51 if (CacheManager.getInstance().updateEdit(cacheBase)) { 52 log.error("成功获得编辑状态"); 53 /*为了模拟并发,我们采用id,保证唯一的数据查看到底谁写入成功*/ 54 cacheBase.setName(GlobalUtil.getId() + ""); 55 CacheManager.getInstance().updateCacheBase(cacheBase); 56 log.error("modelTest.getName()=" + cacheBase.getName()); 57 } else { 58 log.error("获取编辑状态失败"); 59 } 60 } 61 } catch (Exception e) { 62 log.error("更新数据异常", e); 63 } 64 }).start(); 65 } 66 67 } 68 69 }
在mainmanager类main函数测试里面加入3个线程模拟并发状态
正常添加的测试数据
1 [04-07 13:50:50:514:ERROR:MainManager.main():23] 创建数据库,创建数据表结构 2 [04-07 13:50:50:937:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.model.ModelTest 字段:log is transient or static or final; 3 [04-07 13:50:50:952:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.cache.CacheBase 字段:edit is transient or static or final; 4 [04-07 13:50:50:952:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.cache.CacheBase 字段:versionId is transient or static or final; 5 [04-07 13:50:50:952:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.cache.CacheBase 字段:createTime is transient or static or final; 6 [04-07 13:50:50:952:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.cache.CacheBase 字段:lastGetCacheTime is transient or static or final; 7 [04-07 13:51:37:591:ERROR:MainManager.main():41] modelTest.getId()=7040713505000100000 8 [04-07 13:51:45:392:ERROR:MainManager.lambda$main$0():49] 成功获得数据 9 [04-07 13:51:45:392:ERROR:MainManager.lambda$main$0():49] 成功获得数据 10 [04-07 13:51:45:392:ERROR:MainManager.lambda$main$0():49] 成功获得数据 11 [04-07 13:51:45:392:ERROR:MainManager.lambda$main$0():52] 成功获得编辑状态 12 [04-07 13:51:45:392:ERROR:MainManager.lambda$main$0():58] 获取编辑状态失败 13 [04-07 13:51:45:392:ERROR:MainManager.lambda$main$0():58] 获取编辑状态失败 14 [04-07 13:51:45:428:ERROR:CacheManager.updateCacheBase():101] 数据已修改,最新版号:2 15 [04-07 13:51:45:428:ERROR:MainManager.lambda$main$0():56] modelTest.getName()=7040713514500100000
修改后的数据;
保证了并发写入、修改的问题,保证了数据的一致性;
实现滑动缓存
在cache包下面建立里CheckCacheTimer定时器类
1 package net.sz.dbserver.cache; 2 3 import java.util.HashMap; 4 import java.util.Map; 5 import net.sz.framework.szlog.SzLogger; 6 import net.sz.framework.szthread.TimerTaskModel; 7 8 /** 9 * 10 * <br> 11 * author 失足程序员<br> 12 * blog http://www.cnblogs.com/ty408/<br> 13 * mail 492794628@qq.com<br> 14 * phone 13882122019<br> 15 */ 16 public class CheckCacheTimer extends TimerTaskModel { 17 18 private static SzLogger log = SzLogger.getLogger(); 19 20 public CheckCacheTimer(int intervalTime) { 21 super(intervalTime); 22 } 23 24 @Override 25 public void run() { 26 /*考虑缓存的清理的都放在这里、当然有很多值的注意细节有待细化*/ 27 HashMap<Long, CacheBase> tmp = new HashMap(CacheManager.getInstance().cacheMap); 28 for (Map.Entry<Long, CacheBase> entry : tmp.entrySet()) { 29 Long key = entry.getKey(); 30 CacheBase value = entry.getValue(); 31 if (!value.isEdit()) { 32 /*如果数据不在编辑状态、且30分钟无访问清理*/ 33 if (System.currentTimeMillis() - value.getLastGetCacheTime() > 30 * 60 * 1000) { 34 synchronized (CacheManager.getInstance().cacheMap) { 35 if (!value.isEdit()) { 36 /*如果数据不在编辑状态、且30分钟无访问清理*/ 37 if (System.currentTimeMillis() - value.getLastGetCacheTime() > 30 * 60 * 1000) { 38 CacheManager.getInstance().cacheMap.remove(value.getId()); 39 } 40 } 41 } 42 } 43 } 44 } 45 } 46 }
在cachemanager类构造函数加入
1 public CacheManager() { 2 /*创建一秒钟检查的定时器*/ 3 ThreadPool.addTimerTask(ThreadPool.GlobalThread, new CheckCacheTimer(1000)); 4 }
滑动缓存就构建完成了,
这里就不在测试了,理论就是这么个理论;思路就是这么个思路;
脱离数据源的单纯缓存器
改造CacheBase类

改造CacheManager类

改造CheckCacheTimer类

脱离了数据源的缓存器;