重构伪代码
This commit is contained in:
ascrutae
parent
32777d72c7
commit
0dadf27be7
|
|
@ -1,30 +1,28 @@
|
|||
package com.a.eye.skywalking.storage.data;
|
||||
|
||||
import com.a.eye.datacarrier.consumer.IConsumer;
|
||||
import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
|
||||
import com.a.eye.skywalking.storage.data.file.DataFileWriter;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaGroup;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexOperator;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexOperatorCache;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexDBConnectorCache;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public class SpanDataConsumer implements IConsumer<SpanData> {
|
||||
|
||||
private IndexOperatorCache cache;
|
||||
private DataFileWriter fileWriter;
|
||||
private IndexDBConnectorCache cache;
|
||||
private DataFileWriter fileWriter;
|
||||
|
||||
@Override
|
||||
public void consume(List<SpanData> data) {
|
||||
List<IndexMetaInfo> indexMetaInfo = fileWriter.write(data);
|
||||
|
||||
Map<Long, List<IndexMetaInfo>> categorizedMetaInfo =
|
||||
IndexMetaInfoCategory.categorizeByDataIndexTime(indexMetaInfo, BlockIndexEngine.newFinder());
|
||||
Iterator<IndexMetaGroup> iterator = fileWriter.write(data).group().iterator();
|
||||
|
||||
for (Map.Entry<Long, List<IndexMetaInfo>> indexEntry : categorizedMetaInfo.entrySet()) {
|
||||
IndexOperator indexOperator = cache.get(indexEntry.getKey());
|
||||
indexOperator.update(indexEntry.getValue());
|
||||
while (iterator.hasNext()) {
|
||||
IndexMetaGroup metaGroup = iterator.next();
|
||||
IndexOperator indexOperator = IndexOperator.newOperator(cache.get(metaGroup.getTimestamp()));
|
||||
indexOperator.update(metaGroup.getMetaInfo());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.skywalking.storage.data.SpanData;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaCollections;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
|
|
@ -13,7 +13,7 @@ public class DataFileWriter {
|
|||
dataFile = DataFilesManager.createNewDataFile();
|
||||
}
|
||||
|
||||
public List<IndexMetaInfo> write(List<SpanData> spanData) {
|
||||
public IndexMetaCollections write(List<SpanData> spanData) {
|
||||
|
||||
if (dataFile.overLimitLength()) {
|
||||
dataFile = DataFilesManager.createNewDataFile();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,30 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class IndexDBConnector {
|
||||
|
||||
private long timestamp;
|
||||
|
||||
public IndexDBConnector(long timestamp) {
|
||||
|
||||
}
|
||||
|
||||
private void validate() {
|
||||
|
||||
}
|
||||
|
||||
private void createTable() {
|
||||
|
||||
}
|
||||
|
||||
private void createIndex() {
|
||||
|
||||
}
|
||||
|
||||
|
||||
public long getTimestamp() {
|
||||
return timestamp;
|
||||
}
|
||||
}
|
||||
|
|
@ -6,21 +6,21 @@ import java.util.concurrent.locks.Lock;
|
|||
import java.util.concurrent.locks.ReadWriteLock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
|
||||
public class IndexOperatorCache {
|
||||
public class IndexDBConnectorCache {
|
||||
|
||||
private static final int MAX_CACHE_SIZE = 5;
|
||||
|
||||
private LRUCache<Long, IndexOperator> cachedOperators;
|
||||
private LRUCache<Long, IndexDBConnector> cachedOperators;
|
||||
|
||||
public IndexOperatorCache() {
|
||||
cachedOperators = new LRUCache<Long, IndexOperator>(MAX_CACHE_SIZE);
|
||||
public IndexDBConnectorCache() {
|
||||
cachedOperators = new LRUCache<Long, IndexDBConnector>(MAX_CACHE_SIZE);
|
||||
}
|
||||
|
||||
public IndexOperator get(long timestamp) {
|
||||
public IndexDBConnector get(long timestamp) {
|
||||
return cachedOperators.get(timestamp);
|
||||
}
|
||||
|
||||
public void updateCache(long timestamp, IndexOperator operator) {
|
||||
public void updateCache(long timestamp, IndexDBConnector operator) {
|
||||
cachedOperators.put(timestamp, operator);
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
|
||||
import com.a.eye.skywalking.storage.block.index.BlockFinder;
|
||||
import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class IndexMetaCollections {
|
||||
|
||||
private List<IndexMetaInfo> metaInfo;
|
||||
private BlockFinder finder = BlockIndexEngine.newFinder();
|
||||
|
||||
public List<IndexMetaGroup> group() {
|
||||
List<IndexMetaGroup> indexMetaGroups = new ArrayList<IndexMetaGroup>();
|
||||
for (IndexMetaInfo info : metaInfo) {
|
||||
long timestamp = finder.find(info.getStartTime());
|
||||
|
||||
int index = indexMetaGroups.indexOf(new IndexMetaGroup(timestamp));
|
||||
IndexMetaGroup metaGroup;
|
||||
|
||||
if (index == -1) {
|
||||
metaGroup = new IndexMetaGroup(timestamp);
|
||||
indexMetaGroups.add(metaGroup);
|
||||
} else {
|
||||
metaGroup = indexMetaGroups.get(index);
|
||||
}
|
||||
|
||||
metaGroup.addIndexMetaInfo(info);
|
||||
}
|
||||
|
||||
return indexMetaGroups;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class IndexMetaGroup {
|
||||
|
||||
private long timestamp;
|
||||
|
||||
private List<IndexMetaInfo> metaInfo;
|
||||
|
||||
public IndexMetaGroup(long timestamp) {
|
||||
this.timestamp = timestamp;
|
||||
metaInfo = new ArrayList<IndexMetaInfo>();
|
||||
}
|
||||
|
||||
public long getTimestamp() {
|
||||
return timestamp;
|
||||
}
|
||||
|
||||
public List<IndexMetaInfo> getMetaInfo() {
|
||||
return metaInfo;
|
||||
}
|
||||
|
||||
public void addIndexMetaInfo(IndexMetaInfo info) {
|
||||
this.metaInfo.add(info);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o)
|
||||
return true;
|
||||
if (o == null || getClass() != o.getClass())
|
||||
return false;
|
||||
|
||||
IndexMetaGroup that = (IndexMetaGroup) o;
|
||||
|
||||
return timestamp == that.timestamp;
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return (int) (timestamp ^ (timestamp >>> 32));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,18 +1,16 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import static com.a.eye.skywalking.storage.config.Config.DataIndex.TABLE_NAME;
|
||||
|
||||
public class IndexOperator {
|
||||
|
||||
private Connection connection;
|
||||
private long timestamp;
|
||||
|
||||
private IndexOperator(long timestamp) {
|
||||
private IndexDBConnector connector;
|
||||
private long timestamp;
|
||||
|
||||
private IndexOperator(IndexDBConnector connector) {
|
||||
this.connector = connector;
|
||||
timestamp = connector.getTimestamp();
|
||||
}
|
||||
|
||||
public List<IndexMetaInfo> find(String taceId) {
|
||||
|
|
@ -24,31 +22,16 @@ public class IndexOperator {
|
|||
|
||||
}
|
||||
|
||||
private Connection getConnection() {
|
||||
return connection;
|
||||
private IndexDBConnector getConnector() {
|
||||
return connector;
|
||||
}
|
||||
|
||||
public static class Builder {
|
||||
private IndexOperator operator;
|
||||
private IndexOperatorHelper indexOperatorHelper;
|
||||
|
||||
private Builder(long timestamp) {
|
||||
operator = new IndexOperator(timestamp);
|
||||
indexOperatorHelper = new IndexOperatorHelper(operator.getConnection());
|
||||
}
|
||||
|
||||
public static Builder newBuilder(long timestamp) {
|
||||
return new Builder(timestamp);
|
||||
}
|
||||
|
||||
public IndexOperator build() {
|
||||
if (indexOperatorHelper.validateIsReady(TABLE_NAME)) {
|
||||
indexOperatorHelper.maintain();
|
||||
}
|
||||
|
||||
return operator;
|
||||
}
|
||||
|
||||
public static IndexOperator newOperator(long timestamp) {
|
||||
return newOperator(new IndexDBConnector(timestamp));
|
||||
}
|
||||
|
||||
public static IndexOperator newOperator(IndexDBConnector indexDBConnector) {
|
||||
return new IndexOperator(indexDBConnector);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,20 +0,0 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class IndexOperatorFactory {
|
||||
|
||||
private static IndexOperatorCache operatorCache;
|
||||
|
||||
public static IndexOperator getIndexOperator(long timestamp) {
|
||||
IndexOperator operator = operatorCache.get(timestamp);
|
||||
|
||||
if (operator == null) {
|
||||
operator = IndexOperator.Builder.newBuilder(timestamp).build();
|
||||
operatorCache.updateCache(timestamp, operator);
|
||||
}
|
||||
|
||||
return operator;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue