官方的解釋這個(gè)類為：

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

簡(jiǎn)單來(lái)理解就是這個(gè)類每次讀寫都會(huì)將一個(gè)Map<Object, Object>的對(duì)象序列化存儲(chǔ)到磁盤中，讀的時(shí)候?qū)⑵浞葱蛄谢?/p>

構(gòu)造函數(shù)指定的參數(shù)就是你在磁盤中存儲(chǔ)的目錄，同時(shí)也作為VersionedStore的構(gòu)造函數(shù)的參數(shù)。

這些文件在目錄中是以一個(gè)long類型的id進(jìn)行命名

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

snapshot函數(shù)，找到最近的版本，將其反序列化

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會(huì)將其寫入到磁盤中}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);}

可以看到，基本暴露的接口都通過(guò)synchronized關(guān)鍵字來(lái)保證串行化的操作，同時(shí)多次調(diào)用了以下的persist方法，

private void persist(Map<Object, Object> val, boolean cleanup)throws IOException {byte[] toWrite = Utils.serialize(val);String newPath = _vs.createVersion(); //創(chuàng)建一個(gè)新的版本號(hào)FileUtils.writeByteArrayToFile(new File(newPath), toWrite);_vs.succeedVersion(newPath); //如果寫入成功，那么會(huì)生成 id.version 文件來(lái)聲明該文件寫入成功if (cleanup)_vs.cleanup(4); //默認(rèn)保留4個(gè)版本}

接下來(lái)看看VersionedStore這個(gè)類，它是進(jìn)行實(shí)際存儲(chǔ)操作的類，提供了接口給LocalState

public void succeedVersion(String path) throws IOException {long version = validateAndGetVersion(path); //驗(yàn)證一下這個(gè)文件是否存在// should rewrite this to do a file move createNewFile(tokenPath(version)); //創(chuàng)建對(duì)應(yīng)的 id.version 文件說(shuō)明寫入成功}

path的值是一個(gè)long類型的id，表示對(duì)應(yīng)的文件

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

//解析出版本號(hào)，如果以.version結(jié)尾的，去掉.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)); //創(chuàng)建對(duì)應(yīng)的 id.version 文件說(shuō)明寫入成功

token file就是一種標(biāo)志文件，用于標(biāo)志對(duì)應(yīng)的文件已經(jīng)寫入成功，以.version 結(jié)尾

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函數(shù)，保留versionsToKeep版本，清除其他的版本

public void cleanup(int versionsToKeep) throws IOException {List<Long> versions = getAllVersions(); //獲取所有的版本，這個(gè)返回的是以倒序排列的，最新的版本在最前面if (versionsToKeep >= 0) {versions = versions.subList(0,Math.min(versions.size(), versionsToKeep)); //所以可以用subList來(lái)得到需要的版本}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結(jié)尾的文件，也就是說(shuō)所有寫入成功的文件，不包括某些還沒(méi)寫成功的文件

/*** 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)); //驗(yàn)證該文件是否存在}}Collections.sort(ret);Collections.reverse(ret); //逆序排列return ret;}

刪除對(duì)應(yīng)的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);}}

在最開(kāi)始的地方，snapshot()函數(shù)調(diào)用了?mostRecentVersionPath() 來(lái)獲取最近的版本，也就是調(diào)用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號(hào)的話，可以看到是取出了比這個(gè)version號(hào)小的最大的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;}

?

轉(zhuǎn)載于:https://www.cnblogs.com/longshaohang/p/3893264.html

總結(jié)

以上是生活随笔為你收集整理的Storm中的LocalState 代码解析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。