官方的解釋這個類為：

/*** 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小的最大versionreturn v;}return null;}

?

轉載于:https://www.cnblogs.com/longshaohang/p/3893264.html

總結

以上是生活随笔為你收集整理的Storm中的LocalState 代码解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。