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 类

 1 package net.sz.dbserver.model;
 2
 3 import javax.persistence.Id;
 4 import net.sz.framework.szlog.SzLogger;
 5
 6 /**
 7  *
 8  * <br>
 9  * author 失足程序员<br>
10  * blog http://www.cnblogs.com/ty408/<br>
11  * mail [email protected]<br>
12  * phone 13882122019<br>
13  */
14 public class ModelTest {
15
16     private static SzLogger log = SzLogger.getLogger();
17
18     /*主键ID*/
19     @Id
20     private long Id;
21     /*内容*/
22     private String name;
23
24
25     public ModelTest() {
26     }
27
28     public long getId() {
29         return Id;
30     }
31
32     public void setId(long Id) {
33         this.Id = Id;
34     }
35
36     public String getName() {
37         return name;
38     }
39
40     public void setName(String name) {
41         this.name = name;
42     }
43
44 }

然后在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 [email protected]<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 package net.sz.dbserver.main;
 2
 3 import net.sz.dbserver.db.DBManager;
 4 import net.sz.framework.szlog.SzLogger;
 5
 6 /**
 7  *
 8  * <br>
 9  * author 失足程序员<br>
10  * blog http://www.cnblogs.com/ty408/<br>
11  * mail [email protected]<br>
12  * phone 13882122019<br>
13  */
14 public class MainManager {
15
16     private static SzLogger log = SzLogger.getLogger();
17
18     public static void main(String[] args) {
19         log.error("创建数据库,创建数据表结构");
20         DBManager.getInstance().checkTables();
21     }
22
23 }

以上代码我们完成了数据库文件和数据表的创建

 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         }    

输出

 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 11:13:17:904:ERROR:MainManager.main():21] 创建数据库,创建数据表结构
10 [04-07 11:13:18:203:ERROR:Dao.getColumns():532] 类:net.sz.dbserver.model.ModelTest 字段:log is transient or static or final;
11 [04-07 11:13:18:215:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
12 [04-07 11:13:18:216:ERROR:SqliteDaoImpl.existsColumn():130] 数据库:null 表:ModelTest 映射数据库字段:ModelTest 检查结果:已存在,将不会修改
13 [04-07 11:13:18:216:ERROR:SqliteDaoImpl.createTable():168] 表:ModelTest 字段:Id 映射数据库字段:Id 存在,将不会修改,
14 [04-07 11:13:18:216:ERROR:SqliteDaoImpl.existsColumn():130] 数据库:null 表:ModelTest 映射数据库字段:ModelTest 检查结果:已存在,将不会修改
15 [04-07 11:13:18:216:ERROR:SqliteDaoImpl.createTable():168] 表:ModelTest 字段:name 映射数据库字段:name 存在,将不会修改,
16 [04-07 11:13:18:245:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
17 [04-07 11:13:18:245:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
18 [04-07 11:13:18:246:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
19 [04-07 11:13:18:272:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
20 [04-07 11:13:18:272:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
21 [04-07 11:13:18:273:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
22 [04-07 11:13:18:295:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
23 [04-07 11:13:18:296:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
24 [04-07 11:13:18:297:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
25 [04-07 11:13:18:319:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
26 [04-07 11:13:18:319:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
27 [04-07 11:13:18:320:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
28 [04-07 11:13:18:343:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
29 [04-07 11:13:18:343:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
30 [04-07 11:13:18:344:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
31 [04-07 11:13:18:368:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
32 [04-07 11:13:18:368:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
33 [04-07 11:13:18:369:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
34 [04-07 11:13:18:391:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
35 [04-07 11:13:18:391:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
36 [04-07 11:13:18:392:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
37 [04-07 11:13:18:415:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
38 [04-07 11:13:18:415:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
39 [04-07 11:13:18:416:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
40 [04-07 11:13:18:438:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
41 [04-07 11:13:18:439:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
42 [04-07 11:13:18:440:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1
43 [04-07 11:13:18:461:ERROR:SqliteDaoImpl.existsTable():110] 表:ModelTest 检查结果: 已存在
44 [04-07 11:13:18:462:ERROR:Dao.insert():1023] 执行 insert into `ModelTest` (`Id`, `name`) values (?, ? ) 添加数据 表:ModelTest
45 [04-07 11:13:18:463:ERROR:Dao.insert():1067] 执行 [email protected] 添加数据 表:ModelTest 结果 影响行数:1

重构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 [email protected]<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 [email protected]<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 [email protected]", 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 [email protected]<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 [email protected]<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类

  1 package net.sz.net.sz.framework.caches;
  2
  3 import net.sz.framework.util.AtomInteger;
  4
  5 /**
  6  *
  7  * <br>
  8  * author 失足程序员<br>
  9  * blog http://www.cnblogs.com/ty408/<br>
 10  * mail [email protected]<br>
 11  * phone 13882122019<br>
 12  */
 13 public abstract class CacheBase {
 14
 15     /*编辑状态 */
 16     private volatile transient boolean edit;
 17     /*版本号 */
 18     private volatile transient AtomInteger versionId;
 19     /*创建时间*/
 20     private volatile transient long createTime;
 21     /*最后获取缓存时间*/
 22     private volatile transient long lastGetCacheTime;
 23     /*true 表示是滑动缓存*/
 24     private volatile transient boolean slide;
 25     /*清理时间*/
 26     private volatile transient long clearTime;
 27
 28     public CacheBase() {
 29     }
 30
 31     /**
 32      * 创建
 33      * @param slide
 34      * @param clearTime
 35      */
 36     protected CacheBase(boolean slide, long clearTime) {
 37         this.slide                      = slide;
 38         this.clearTime                  = clearTime;
 39     }
 40
 41     /**
 42      *
 43      */
 44     protected void createCache(){
 45         this.edit                       = false;
 46         this.versionId                  = new AtomInteger(1);
 47         this.createTime                 = System.currentTimeMillis();
 48         this.lastGetCacheTime           = System.currentTimeMillis();
 49     }
 50
 51         /**
 52      *
 53      */
 54     protected void copyCache(CacheBase tmp){
 55         this.edit                       = tmp.edit;
 56         this.versionId                  = tmp.getVersionId();
 57         this.createTime                 = tmp.getClearTime();
 58         this.lastGetCacheTime           = System.currentTimeMillis();
 59     }
 60
 61     public boolean isEdit() {
 62         return edit;
 63     }
 64
 65     public void setEdit(boolean edit) {
 66         this.edit = edit;
 67     }
 68
 69     public AtomInteger getVersionId() {
 70         return versionId;
 71     }
 72
 73     public void setVersionId(AtomInteger versionId) {
 74         this.versionId = versionId;
 75     }
 76
 77     public long getCreateTime() {
 78         return createTime;
 79     }
 80
 81     public void setCreateTime(long createTime) {
 82         this.createTime = createTime;
 83     }
 84
 85     public long getLastGetCacheTime() {
 86         return lastGetCacheTime;
 87     }
 88
 89     public void setLastGetCacheTime(long lastGetCacheTime) {
 90         this.lastGetCacheTime = lastGetCacheTime;
 91     }
 92
 93     public boolean isSlide() {
 94         return slide;
 95     }
 96
 97     public void setSlide(boolean slide) {
 98         this.slide = slide;
 99     }
100
101     public long getClearTime() {
102         return clearTime;
103     }
104
105     public void setClearTime(long clearTime) {
106         this.clearTime = clearTime;
107     }
108
109 }

改造CacheManager类

  1 package net.sz.net.sz.framework.caches;
  2
  3 import java.util.concurrent.ConcurrentHashMap;
  4 import net.sz.framework.szlog.SzLogger;
  5 import net.sz.framework.szthread.ThreadPool;
  6
  7 /**
  8  *
  9  * <br>
 10  * author 失足程序员<br>
 11  * blog http://www.cnblogs.com/ty408/<br>
 12  * mail [email protected]<br>
 13  * phone 13882122019<br>
 14  */
 15 public class CacheManager {
 16
 17     private static SzLogger log = SzLogger.getLogger();
 18
 19     /*缓存集合*/
 20     final ConcurrentHashMap<String, CacheBase> cacheMap = new ConcurrentHashMap<>();
 21
 22     public CacheManager() {
 23         /*创建一秒钟检查的定时器*/
 24         ThreadPool.addTimerTask(ThreadPool.GlobalThread, new CheckCacheTimer(1000, this));
 25     }
 26
 27     /**
 28      * 默认30分钟清理的滑动缓存、如果存在缓存键、将不再添加
 29      *
 30      * @param key
 31      * @param object
 32      * @return
 33      */
 34     public boolean addCache(String key, CacheBase object) {
 35         return addCache(key, object, 30 * 60 * 1000);
 36     }
 37
 38     /**
 39      * 默认滑动缓存、如果存在缓存键、将不再添加
 40      *
 41      * @param key
 42      * @param object
 43      * @param clearTime 滑动缓存的清理时间
 44      * @return
 45      */
 46     public boolean addCache(String key, CacheBase object, long clearTime) {
 47         return addCache(key, object, clearTime, true);
 48     }
 49
 50     /**
 51      * 默认滑动缓存、如果存在缓存键、将不再添加
 52      *
 53      * @param key
 54      * @param object
 55      * @param clearTime 清理缓存的间隔时间
 56      * @param isSlide true表示滑动缓存,
 57      * @return
 58      */
 59     public boolean addCache(String key, CacheBase object, long clearTime, boolean isSlide) {
 60         return addCache(key, object, clearTime, isSlide, false);
 61     }
 62
 63     /**
 64      *
 65      * @param key
 66      * @param object
 67      * @param clearTime 清理缓存的间隔时间
 68      * @param isSlide true表示滑动缓存,
 69      * @param put true 表示强制添加数据集合,及已经存在键数据
 70      * @return
 71      */
 72     public boolean addCache(String key, CacheBase object, long clearTime, boolean isSlide, boolean put) {
 73         if (put) {
 74             object.createCache();
 75             cacheMap.put(key, object);
 76             if (log.isDebugEnabled()) {
 77                 log.debug("强制添加缓存键=" + key);
 78             }
 79             return true;
 80         } else if (!cacheMap.containsKey(key)) {
 81             /*加入同步操作*/
 82             synchronized (key) {
 83                 if (!cacheMap.containsKey(key)) {
 84                     object.createCache();
 85                     cacheMap.put(key, object);
 86                     return true;
 87                 } else {
 88                     if (log.isDebugEnabled()) {
 89                         log.debug("数据已经添加,不能再次添加");
 90                     }
 91                 }
 92             }
 93         } else {
 94             if (log.isDebugEnabled()) {
 95                 log.debug("数据已经添加,不能再次添加");
 96             }
 97         }
 98         return false;
 99     }
100
101     public boolean removeCache(String key) {
102         cacheMap.remove(key);
103         return true;
104     }
105
106     public CacheBase getCache(String key) {
107         return getCache(key, CacheBase.class);
108     }
109
110     /**
111      * 获取一条数据,这里我只是测试,提供思路,
112      * <br>
113      * 所以不会去考虑list等情况;
114      * <br>
115      * 需要的话可以自行修改
116      *
117      * @param <T>
118      * @param clazz
119      * @param key
120      * @return
121      */
122     public <T extends CacheBase> T getCache(String key, Class<T> clazz) {
123         CacheBase cacheBase = null;
124         cacheBase = cacheMap.get(key);
125         if (cacheBase != null) {
126             /*更新最后获取缓存的时间*/
127             cacheBase.setLastGetCacheTime(System.currentTimeMillis());
128         }
129         return (T) cacheBase;
130     }
131
132     /**
133      * 更新缓存数据同时更新数据库数据
134      *
135      * @param key
136      * @param object
137      * @return
138      */
139     public boolean updateCacheBase(String key, CacheBase object) {
140         if (object == null) {
141             throw new UnsupportedOperationException("参数 object 为 null");
142         }
143         CacheBase cacheBase = null;
144         cacheBase = cacheMap.get(key);
145         /*理论上,控制得当这里是不可能为空的*/
146         if (cacheBase != null) {
147             /*理论上是能绝对同步的,你也可以稍加修改*/
148             synchronized (key) {
149                 /*验证编辑状态和版号,保证写入数据是绝对正确的*/
150                 if (cacheBase.getVersionId() == object.getVersionId()) {
151                     /*拷贝最新数据操作*/
152                     cacheMap.put(key, object);
153                     /*保证写入数据库后进行修改 对版本号进行加一操作*/
154                     object.getVersionId().changeZero(1);
155                     /*设置最新的最后访问时间*/
156                     object.setLastGetCacheTime(System.currentTimeMillis());
157                     /*修改编辑状态*/
158                     object.setEdit(false);
159                     if (log.isDebugEnabled()) {
160                         log.debug("数据已修改,最新版号:" + object.getVersionId());
161                     }
162                     return true;
163                 } else {
164                     if (log.isDebugEnabled()) {
165                         log.debug("版本已经修改无法进行更新操作");
166                     }
167                     throw new UnsupportedOperationException("版本已经修改无法进行更新操作");
168                 }
169             }
170         } else {
171             if (log.isDebugEnabled()) {
172                 log.debug("缓存不存在无法修改数据");
173             }
174             throw new UnsupportedOperationException("缓存不存在无法修改数据");
175         }
176     }
177
178     /**
179      * 获取独占编辑状态
180      *
181      * @param key
182      * @return
183      */
184     public boolean updateEdit(String key) {
185         CacheBase cacheBase = null;
186         cacheBase = cacheMap.get(key);
187         if (cacheBase == null) {
188             throw new UnsupportedOperationException("未找到数据源");
189         }
190         return updateEdit(key, cacheBase);
191     }
192
193     /**
194      * 获取独占编辑状态
195      *
196      * @param key
197      * @param cacheBase
198      * @return
199      */
200     public boolean updateEdit(String key, CacheBase cacheBase) {
201         if (cacheBase == null) {
202             throw new UnsupportedOperationException("参数 cacheBase 为 null");
203         }
204         if (!cacheBase.isEdit()) {
205             synchronized (key) {
206                 if (!cacheBase.isEdit()) {
207                     /*同步后依然需要双重判定*/
208                     cacheBase.setEdit(true);
209                     /*设置最新的最后访问时间*/
210                     cacheBase.setLastGetCacheTime(System.currentTimeMillis());
211                     return true;
212                 }
213             }
214         }
215         return false;
216     }
217
218 }

改造CheckCacheTimer类

 1 package net.sz.net.sz.framework.caches;
 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 [email protected]<br>
14  * phone 13882122019<br>
15  */
16 public class CheckCacheTimer extends TimerTaskModel {
17
18     private static SzLogger log = SzLogger.getLogger();
19     private final CacheManager cacheManager;
20
21     public CheckCacheTimer(int intervalTime, CacheManager cacheManager) {
22         super(intervalTime);
23         this.cacheManager = cacheManager;
24     }
25
26     @Override
27     public void run() {
28         /*考虑缓存的清理的都放在这里、当然有很多值的注意细节有待细化*/
29         HashMap<String, CacheBase> tmp = new HashMap(this.cacheManager.cacheMap);
30         for (Map.Entry<String, CacheBase> entry : tmp.entrySet()) {
31             String key = entry.getKey();
32             CacheBase value = entry.getValue();
33             /*理论上,这里是能够保证绝对缓存,同步*/
34             if (!value.isEdit()) {
35                 /*滑动缓存清理*/
36                 if (value.isSlide() && System.currentTimeMillis() - value.getLastGetCacheTime() < value.getClearTime()) {
37                     continue;
38                 }
39                 /*固定缓存清理*/
40                 if (!value.isSlide() && System.currentTimeMillis() - value.getCreateTime() < value.getClearTime()) {
41                     continue;
42                 }
43                 this.cacheManager.removeCache(key);
44             }
45         }
46     }
47 }

脱离了数据源的缓存器;

求大神指教了;如果觉得可以点个推荐;

觉得不好请手下留情不要点击反对哦,

时间: 2024-10-06 04:00:44

net.sz.framework 框架 轻松搭建数据服务中心----读写分离数据一致性,滑动缓存的相关文章

net.sz.framework 框架 轻松搭建服务---让你更专注逻辑功能---初探

前言 在之前的文章中,讲解过 threadmodel,socket tcp ,socket http,log,astart ,scripts: 都是分片讲解,从今天开始,将带大家,一窥 net.sz.framework 框架: net.sz.framework 框架分为java版本和C#.net 版本,两种语言版本保持高一致性: net.sz.framework 能做什么呢? net.sz.framework 是经过多年实践和研究,包括在线项目验证的底层框架,致力于解决框架问题: 让你更轻松,愉

基于SQL Server搭建主从复制实现读写分离实战演练

一.课程介绍 读写分离(主从同步)从字面意思就可以理解,就是把对数据库的读操作和写操作分离开.读写分离在网站发展初期可以一定程度上缓解读写并发时产生锁的问题,将读写压力分担到多台服务器上.读写分离的基本原理是让主数据库处理事务性增.改.删操作(INSERT.UPDATE.DELETE),而从数据库处理SELECT查询操作.随着系统的业务量不断增长数据多了之后,对数据库的读.写就会很多.分库减少单台数据库的压力.本文以MS SQL Server数据库为例主库负责写数据.读数据.读库仅负责读数据.写

CentOS 6.8搭建mysql-proxy实现读写分离

CentOS 6.8搭建mysql-proxy实现读写分离: 下载网址:http://ftp.ntu.edu.tw/MySQL/Downloads/MySQL-Proxy 注:1.10.0.1.2为mysql-proxy服务器ip2.效果是1个写,2个读 tar zxfv mysql-proxy-0.8.5-linux-el6-x86-64bit.tar.gz -C /space mv /space/mysql-proxy-0.8.5-linux-el6-x86-64bit /space/mys

超详细搭建Mysql5.5读写分离

Amoeba简介 Amoeba(变形虫)项目,该开源框架于2008年 开始发布一款 Amoeba for Mysql软件.这个软件致力于MySQL的分布式数据库前端代理层,它主要在应用层访问MySQL的 时候充当SQL路由功能,专注于分布式数据库代理层(Database Proxy)开发.座落与 Client.DB Server(s)之间,对客户端透明.具有负载均衡.高可用性.SQL 过滤.读写分离.可路由相关的到目标数据库.可并发请求多台数据库合并结果. 通过Amoeba你能够完成多数据源的高

elasticsearch 冷热数据的读写分离

步骤 一.冷热分离集群配置 比如三个机器共六个node的es集群. 每个机器上各挂载一个ssd 和 一个sata.每个机器需要启动两个es进程.每个进程对应不同类型的磁盘. 关键配置: node.max_local_storage_nodes: 2   #允许每个机器启动两个es进程 path.data: /home/centos/es/elasticsearch-2.1.0/data/ssd   #需要显示指定es对应的数据目录 启动命令中需要指定node tag ./elasticsearc

MySQL Proxy实现数据的读写分离

做MySQL读写分离实验之前,先要搭建MySQL主从复制,可以参考我之前的两篇文章: MySQL主从配置:http://msiyuetian.blog.51cto.com/8637744/1697288 部署搭建分层次LAMP架构(最后一章节):http://msiyuetian.blog.51cto.com/8637744/1740236 实现了主从复制后,下面我们进行读写分离的安装与配置.环境说明: Linux发行版本:centos 6.7 x86_64 (三台) Hostname IP 备

Spring和MyBatis实现数据的读写分离

1.Spring实现数据库的读写分离 现在大型的电子商务系统,在数据库层面大都采用读写分离技术,就是一个Master数据库,多个Slave数据库.Master库负责数据更新和实时数据查询,Slave库当然负责非实时数据查询.因为在实际的应用中,数据库都是读多写少(读取数据的频率高,更新数据的频率相对较少),而读取数据通常耗时比较长,占用数据库服务器的CPU较多,从而影响用户体验.我们通常的做法就是把查询从主库中抽取出来,采用多个从库,使用负载均衡,减轻每个从库的查询压力. 采用读写分离技术的目标

深入了解Entity Framework框架及访问数据的几种方式

一.前言 1.Entity Framework概要 Entity Framework是微软以ADO.NET为基础所发展出来的对象关系映射(O/R Mapping)解决方案.该框架曾经为.NET Framework的一部分,但Version 6之后从.NET Framework分离出来,可通过NuGet获取. Entity Framework利用抽象化数据结构的方式,将每个数据库对象都转换成应用程序对象 (Entity),而数据字段都转换为属性 (Property),关系则转换为结合属性 (Ass

在CentOS7上搭建MySQL主从复制与读写分离

MySQL主从复制原理 MySQL的主从复制和MySQL的读写分离两者有着紧密联系,首先要部署主从复制,只有主从复制完成了,才能在此基础上进行数据的读写分离. (1)MySQL支持复制的类型. 1)基于语句的复制.MySQL默认采用基于语句的复制,效率比较高. 2)基于行的复制.把改变的内容复制过去,而不是把命令在从服务器上执行一遍. 3)混合类型的复制.默认采用基于语句的复制,一旦发现基于语句无法精确复制时,就会采用基于行的复制. (2)MySQL复制的工作过程如图所示.1)在每个事务更新数据