官方的解释这个类为：
/** * 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 代码解析,布布扣,bubuko.com
时间： 2024-10-24 19:51:19