官方的解释这个类为:

/**
* A simple, durable, atomic K/V database. *Very inefficient*, should only be
* used for occasional reads/writes. Every read/write hits disk.
*/

简单来理解就是这个类每次读写都会将一个Map<Object, Object>的对象序列化存储到磁盘中,读的时候将其反序列化。

构造函数指定的参数就是你在磁盘中存储的目录,同时也作为VersionedStore的构造函数的参数。

这些文件在目录中是以一个long类型的id进行命名

public LocalState(String backingDir) throws IOException {
_vs = new VersionedStore(backingDir);
}

snapshot函数,找到最近的版本,将其反序列化

    public synchronized Map<Object, Object> snapshot() throws IOException {
int attempts = 0;
while (true) {
String latestPath = _vs.mostRecentVersionPath(); //获取最近的版本
if (latestPath == null)
return new HashMap<Object, Object>();
try {
return (Map<Object, Object>) Utils.deserialize(FileUtils
.readFileToByteArray(new File(latestPath)));
} catch (IOException e) {
attempts++;
if (attempts >= 10) {
throw e;
}
}
}
}
    public Object get(Object key) throws IOException {
return snapshot().get(key);
} public synchronized void put(Object key, Object val) throws IOException {
put(key, val, true);
} public synchronized void put(Object key, Object val, boolean cleanup)
throws IOException {
Map<Object, Object> curr = snapshot();
curr.put(key, val);
persist(curr, cleanup); //persist会将其写入到磁盘中
} public synchronized void remove(Object key) throws IOException {
remove(key, true);
} public synchronized void remove(Object key, boolean cleanup)
throws IOException {
Map<Object, Object> curr = snapshot();
curr.remove(key);
persist(curr, cleanup);
} public synchronized void cleanup(int keepVersions) throws IOException {
_vs.cleanup(keepVersions);
}

可以看到,基本暴露的接口都通过synchronized关键字来保证串行化的操作,同时多次调用了以下的persist方法,

    private void persist(Map<Object, Object> val, boolean cleanup)
throws IOException {
byte[] toWrite = Utils.serialize(val);
String newPath = _vs.createVersion(); //创建一个新的版本号
FileUtils.writeByteArrayToFile(new File(newPath), toWrite);
_vs.succeedVersion(newPath); //如果写入成功,那么会生成 id.version 文件来声明该文件写入成功
if (cleanup)
_vs.cleanup(4); //默认保留4个版本
}

接下来看看VersionedStore这个类,它是进行实际存储操作的类,提供了接口给LocalState

    public void succeedVersion(String path) throws IOException {
long version = validateAndGetVersion(path); //验证一下这个文件是否存在
// should rewrite this to do a file move
createNewFile(tokenPath(version)); //创建对应的 id.version 文件说明写入成功
}

path的值是一个long类型的id,表示对应的文件

    private long validateAndGetVersion(String path) {
Long v = parseVersion(path);
if (v == null)
throw new RuntimeException(path + " is not a valid version");
return v;
}

//解析出版本号,如果以.version结尾的,去掉.version

    private Long parseVersion(String path) {
String name = new File(path).getName();
if (name.endsWith(FINISHED_VERSION_SUFFIX)) {
name = name.substring(0,
name.length() - FINISHED_VERSION_SUFFIX.length());
}
try {
return Long.parseLong(name);
} catch (NumberFormatException e) {
return null;
}
}
 createNewFile(tokenPath(version));   //创建对应的 id.version 文件说明写入成功

token file就是一种标志文件,用于标志对应的文件已经写入成功,以.version 结尾

    private String tokenPath(long version) {
return new File(_root, "" + version + FINISHED_VERSION_SUFFIX)
.getAbsolutePath();
}
    private void createNewFile(String path) throws IOException {
new File(path).createNewFile();
}

cleanup函数,保留versionsToKeep版本,清除其他的版本

    public void cleanup(int versionsToKeep) throws IOException {
List<Long> versions = getAllVersions(); //获取所有的版本,这个返回的是以倒序排列的,最新的版本在最前面
if (versionsToKeep >= 0) {
versions = versions.subList(0,
Math.min(versions.size(), versionsToKeep)); //所以可以用subList来得到需要的版本
}
HashSet<Long> keepers = new HashSet<Long>(versions); //存在HashSet中方便快速存取 for (String p : listDir(_root)) {
Long v = parseVersion(p);
if (v != null && !keepers.contains(v)) {
deleteVersion(v); //删除其他的版本
}
}
}

getAllVersions,注意这里是获取所有以version结尾的文件,也就是说所有写入成功的文件,不包括某些还没写成功的文件

    /**
* Sorted from most recent to oldest
*/
public List<Long> getAllVersions() throws IOException {
List<Long> ret = new ArrayList<Long>();
for (String s : listDir(_root)) { //获取该目录下的所有文件
if (s.endsWith(FINISHED_VERSION_SUFFIX)) {
ret.add(validateAndGetVersion(s)); //验证该文件是否存在
}
}
Collections.sort(ret);
Collections.reverse(ret); //逆序排列
return ret;
}

删除对应的version文件和token文件

    public void deleteVersion(long version) throws IOException {
File versionFile = new File(versionPath(version));
File tokenFile = new File(tokenPath(version)); if (versionFile.exists()) {
FileUtils.forceDelete(versionFile);
}
if (tokenFile.exists()) {
FileUtils.forceDelete(tokenFile);
}
}

在最开始的地方,snapshot()函数调用了 mostRecentVersionPath() 来获取最近的版本,也就是调用getAllVersions,然后拿到最新的version

    public String mostRecentVersionPath() throws IOException {
Long v = mostRecentVersion();
if (v == null)
return null;
return versionPath(v);
}
    public Long mostRecentVersion() throws IOException {
List<Long> all = getAllVersions();
if (all.size() == 0)
return null;
return all.get(0);
}

如果提供了version号的话,可以看到是取出了比这个version号小的最大的version

    public String mostRecentVersionPath(long maxVersion) throws IOException {
Long v = mostRecentVersion(maxVersion);
if (v == null)
return null;
return versionPath(v);
}
    public Long mostRecentVersion(long maxVersion) throws IOException {
List<Long> all = getAllVersions();
for (Long v : all) {
if (v <= maxVersion) //取出比maxVersion小的最大version
return v;
}
return null;
}

Storm中的LocalState 代码解析的更多相关文章

  1. linux内存管理--slab及其代码解析

    Linux内核使用了源自于 Solaris 的一种方法,但是这种方法在嵌入式系统中已经使用了很长时间了,它是将内存作为对象按照大小进行分配,被称为slab高速缓存. 内存管理的目标是提供一种方法,为实 ...

  2. Postgres中postmaster代码解析(中)

    今天我们对postmaster的以下细节进行讨论: backend的启动和client的连接请求的认证 客户端取消查询时的处理 接受pg_ctl的shutdown请求进行shutdown处理 2.与前 ...

  3. Python中sort和sorted函数代码解析

    Python中sort和sorted函数代码解析 本文研究的主要是Python中sort和sorted函数的相关内容,具体如下. 一.sort函数 sort函数是序列的内部函数 函数原型: L.sor ...

  4. 捕捉WPF应用程序中XAML代码解析异常

    原文:捕捉WPF应用程序中XAML代码解析异常 由于WPF应用程序中XAML代码在很多时候是运行时加载处理的.比如DynamicResource,但是在编译或者运行的过程中,编写的XAML代码很可能有 ...

  5. .NET Core中的认证管理解析

    .NET Core中的认证管理解析 0x00 问题来源 在新建.NET Core的Web项目时选择“使用个人用户账户”就可以创建一个带有用户和权限管理的项目,已经准备好了用户注册.登录等很多页面,也可 ...

  6. VBA常用代码解析

    031 删除工作表中的空行 如果需要删除工作表中所有的空行,可以使用下面的代码. Sub DelBlankRow() DimrRow As Long DimLRow As Long Dimi As L ...

  7. Storm中遇到的日志多次重写问题(一)

    业务描述: 统计从kafka spout中读取的数据条数,以及写入redis的数据的条数,写入hdfs的数据条数,写入kafaka的数据条数.并且每过5秒将数据按照json文件的形式写入日志.其中保存 ...

  8. OS中atomic的实现解析

    OS中atomic的实现解析 转自:http://my.oschina.net/majiage/blog/267409    摘要 atomic属性线程安全,会增加一定开销,但有些时候必须自定义ato ...

  9. [nRF51822] 10、基础实验代码解析大全 &#183; 实验15 - RTC

    一.实验内容: 配置NRF51822 的RTC0 的TICK 频率为8Hz,COMPARE0 匹配事件触发周期为3 秒,并使能了TICK 和COMPARE0 中断. TICK 中断中驱动指示灯D1 翻 ...

随机推荐

  1. Maven 配置远程仓库

    最近要用到一个包:spark-1.6.0-cdh5.8.2,在https://mvnrepository.com网站上搜到该包的pom.xml语句.但是看到下面有这样一句话: 该包在图中所述repos ...

  2. net.sf.json 时间格式的转化

    后台代码 //后台代码 response.setCharacterEncoding("UTF-8"); JsonConfig jsonConfig = new JsonConfig ...

  3. Vim经典讲解

    http://blog.csdn.net/niushuai666/article/details/7275406

  4. Delphi 的 7zip 压缩算法

    http://blog.csdn.net/warrially/article/details/8039915

  5. Silverlight中动画的性能浅析

    Silverlight中提供了StoryBoard实现动画,可是StoryBoard的性能实在不敢恭维,特别是动画很大的时候,计算机的CPU和内存的狂增,如此一来性能实在太差,在默认的动画效果中动画实 ...

  6. aspose.word 在书签处插入符号

    doc.Range.Bookmarks["CBJYQQDFS110"].Text = ""; Aspose.Words.DocumentBuilder buil ...

  7. jdbc01

    1.创建对应的数据库以及表 /* SQLyog 企业版 - MySQL GUI v8.14 MySQL - 5.5.32-log : Database - news ***************** ...

  8. 解决Idea无法提示代码、不检查语法的方法

    今天打开Idea做项目的时候,java代码图标出现异常(不是以前的C图标),所有java文件都只有两种颜色,百度查了一下,Idea有一个叫power save mode,在file -> Pow ...

  9. OGG学习笔记05-OGG的版本

    刚接触OGG的时候,很容易被众多的版本搞晕,虽然官方有提供各版本对应认证OS和DB的表格. 个人认为一个比较简单的方式,是直接去edelivery.oracle.com下载OGG,选定一个大版本后,这 ...

  10. GreenDao 使用和数据库升级

    1使用方法 一.添加依赖 在bulid.gradle文件下的dependencies下添加所需依赖   compile 'org.greenrobot:greendao:3.2.2' // add l ...