1)关闭xdb
myXdb.stop();2)xdb.Xdb#close
private final synchronized void close() {Trace.warn("xdb stop begin");this.openFlag = false;mbeans().unregisterAll();UniqName.stop();Engine.getInstance().close();Executor.stop();if (null != this.angelThread) {this.angelThread.shutdown();this.angelThread = null;}if (null != this.checkpoint) {this.checkpoint.shutdown();this.checkpoint = null;}// 重点： 设置关闭缓存. 这个其实是ThreadHelper的关闭// 也就是CheckpointCache继承于ThreadHelperif (null != this.checkpointcache) {this.checkpointcache.shutdown();this.checkpointcache = null;}if (null != this.tables) {this.tables.close();this.tables = null;}MySqlMgr.getInstance().exit();Trace.warn("xdb stop end");}3)xdb.CheckpointCache#run // 既然是一个线程，那就看执行体
public void run() {while(super.isRunning()) {long now = System.currentTimeMillis();this.save(PROCESS_MAX_COUNT);long useTime = System.currentTimeMillis() - now;if (useTime >= (long)PROCESS_TIME) {super.sleepIdle(100L);Trace.info("checkpoint cache process time is " + useTime + "ms!");} else {super.sleepIdle((long)PROCESS_TIME - useTime);}}Trace.warn("final CheckPointCache begin");// 重点: 因此这个才是关闭的本质this.save(2147483647);Trace.warn("final CheckPointCache end");int totalCacheCount = this.totalCacheCount();if (totalCacheCount > 0) {Trace.error("final CheckPointCache left " + totalCacheCount);for(int i = 0; i < 60; ++i) {this.save(2147483647);totalCacheCount = this.totalCacheCount();if (totalCacheCount == 0) {break;}Trace.error("final CheckPointCache i=" + i + " left=" + totalCacheCount);super.sleepIdle(10000L);}}}4)xdb.CheckpointCache#save // 保存
private void save(int maxCount) {// 拿到所有的表Iterator iter = this.cacheTables.values().iterator();while(iter.hasNext()) {// 挨个执行save((CheckpointCacheInfo)iter.next()).save(maxCount);}}5)xdb.CheckpointCacheInfo#save
public void save(int maxCount) {if (this.record.size() != 0) {int initRecordCount = this.record.size();HashSet<String> failedRecordKey = new HashSet();Iterator<Entry<String, Treble<State, OctetsStream, Long>>> iter = this.record.entrySet().iterator();int count = 0;String key;String var10000;if (initRecordCount >= CheckpointCache.PROCESS_BATCH || maxCount == 2147483647) {long now = System.currentTimeMillis();Connection conn = null;try {conn = MySqlMgr.getInstance().getWriteConn();PreparedStatement pstReplace = conn.prepareStatement("REPLACE INTO " + this.tableName + " VALUES(?, ?)");pstReplace.setQueryTimeout(10);PreparedStatement pstRemove = conn.prepareStatement("DELETE FROM " + this.tableName + " WHERE k=?");pstRemove.setQueryTimeout(10);label436:while(true) {while(true) {Entry entry;do {if (!iter.hasNext()) {break label436;}entry = (Entry)iter.next();} while((Long)((Treble)entry.getValue()).third >= now && maxCount < 2147483647);if (this.isZero((String)entry.getKey())) {var10000 = (String)entry.getKey();Trace.error("Key is Zero! Key=" + var10000 + " tableName=" + this.tableName);failedRecordKey.add((String)entry.getKey());} else {if (((Treble)entry.getValue()).first != State.INDB_GET && ((Treble)entry.getValue()).first != State.INDB_ADD && ((Treble)entry.getValue()).first != State.ADD) {if (((Treble)entry.getValue()).first != State.INDB_REMOVE) {continue;}failedRecordKey.add((String)entry.getKey());pstRemove.setString(1, (String)entry.getKey());pstRemove.addBatch();} else {failedRecordKey.add((String)entry.getKey());pstReplace.setString(1, (String)entry.getKey());if (Xdb.getInstance().isJsonFlag()) {pstReplace.setString(2, ((OctetsStream)((Treble)entry.getValue()).second).unmarshalJson());} else {pstReplace.setBinaryStream(2, new ByteArrayInputStream(((OctetsStream)((Treble)entry.getValue()).second).array()), ((OctetsStream)((Treble)entry.getValue()).second).size());}pstReplace.addBatch();}++count;if (count >= maxCount) {break label436;}}}}if (count > 0) {pstReplace.executeBatch();pstRemove.executeBatch();conn.commit();}pstReplace.close();pstRemove.close();Iterator var34 = failedRecordKey.iterator();label410:while(true) {do {if (!var34.hasNext()) {failedRecordKey.clear();long useTime = System.currentTimeMillis() - now;if (useTime > (long)MySqlMgr.MS_DEBUG && Trace.isDebugEnabled()) {Trace.debug("Execute Batch Use " + useTime + " tableName is " + this.tableName);}if (useTime > (long)MySqlMgr.MS_INFO * 2L) {Trace.info("Execute Batch Use " + useTime + " tableName is " + this.tableName);}break label410;}key = (String)var34.next();} while((Long)((Treble)this.record.get(key)).third >= now && maxCount < 2147483647);this.record.remove(key);}} catch (Exception var25) {if (conn != null) {try {conn.rollback();Trace.error("save exec Batch Rollback Success " + this.tableName);} catch (Exception var24) {Trace.error("save exec Batch Rollback Exception " + this.tableName, var24);}}Trace.error("save exec Batch ReplaceSQL Exception " + this.tableName, var25);} finally {if (conn != null) {try {conn.close();} catch (SQLException var22) {Trace.error("write conn close", var22);}}}}if (initRecordCount < CheckpointCache.PROCESS_BATCH || failedRecordKey.size() > 0 || maxCount == 2147483647) {int countInit = count;long now = System.currentTimeMillis();iter = this.record.entrySet().iterator();label389:while(true) {while(true) {boolean bSuccess;Entry entry;do {if (!iter.hasNext()) {break label389;}bSuccess = true;entry = (Entry)iter.next();} while((Long)((Treble)entry.getValue()).third >= now && maxCount < 2147483647);if (this.isZero((String)entry.getKey())) {var10000 = (String)entry.getKey();Trace.error("Key is Zero! Key=" + var10000 + " tableName=" + this.tableName);failedRecordKey.add((String)entry.getKey());} else {switch((State)((Treble)entry.getValue()).first) {case INDB_REMOVE:var10000 = this.tableName;key = "DELETE FROM " + var10000 + " WHERE k='" + (String)entry.getKey() + "'";if (MySqlMgr.getInstance().execUpdate(key, (OctetsStream)null) < 0 && MySqlMgr.getInstance().execUpdate(key, (OctetsStream)null) < 0) {failedRecordKey.add((String)entry.getKey());bSuccess = false;}case REMOVE:default:break;case INDB_GET:case INDB_ADD:case ADD:var10000 = this.tableName;String replaceSql = "REPLACE INTO " + var10000 + " VALUES('" + (String)entry.getKey() + "', ?)";if (MySqlMgr.getInstance().execUpdate(replaceSql, (OctetsStream)((Treble)entry.getValue()).second) < 0 && MySqlMgr.getInstance().execUpdate(replaceSql, (OctetsStream)((Treble)entry.getValue()).second) < 0) {failedRecordKey.add((String)entry.getKey());bSuccess = false;}}if (bSuccess && ((Long)((Treble)this.record.get(entry.getKey())).third < now || maxCount == 2147483647)) {iter.remove();}if (bSuccess) {++count;if (count - countInit >= CheckpointCache.PROCESS_SINGLE && maxCount < CheckpointCache.PROCESS_MAX_COUNT) {break label389;}}}}}}if (failedRecordKey.size() > 0) {var10000 = this.tableName;Trace.error(var10000 + " success:" + count + " failed:" + failedRecordKey.size() + " left:" + this.record.size());Iterator var28 = failedRecordKey.iterator();while(var28.hasNext()) {String key = (String)var28.next();Treble<State, OctetsStream, Long> info = (Treble)this.record.get(key);if (info != null && info.second != null) {try {String jsonStr = ((OctetsStream)info.second).unmarshalJson();if (jsonStr.length() > 1000) {Trace.error(this.tableName + " key=" + key + " json size=" + jsonStr.length());} else {Trace.error(this.tableName + " key=" + key + " json=" + jsonStr);}} catch (MarshalException var23) {Trace.error(this.tableName + " key=" + key, var23);}}if (MySqlMgr.failedClearMap.get(this.tableName + key) != null) {this.record.remove(key);Trace.error(this.tableName + " key=" + key + " is removed!");}}} else {Trace.info(this.tableName + " success:" + count + " left:" + this.record.size());}}}理解：
其实就是找到key和value, value是json。
而且是批量处理。
字段是删除还是别的，如果是删除会执行delete操作删除数据。