TinkerGraph源码学习(一)

因为图服务的整个流程最近才得以完善, 市面上开源的图计算或图存储一般实现都很有些复杂(比如JanusGraph,GraphX) ,就像自己动手写OS/虚拟机/Parser一样, 我觉得很多时候如果能自己动手去写一个图系统. 那么对整个图和Tinkerpop的了解我相信会好很多.

今天的主人公就是Tinkerpop的作者自己写的In-memory 图系统—-TinkerGraph ,虽然是基于内存实现,但是提供了持久化的选择, 以及图计算+图存储的基本实现. 觉得很适合作为学习项目.

0x00.代码结构

2.21更新: TinkerGraph后面又小更新了一下, 大家自行拉取官方最新版, 细节变动不影响整体阅读~

首先要知道, Tinkerpop(TP)自带的**七个图数据结构(接口)**需要全部实现, 所以结构这基本就是实现图结构后的对应类. TinkerGraph(TG)如下图所示:

• 处理加工
  1. 图计算
  2. 图遍历
• 数据结构

0x01.structure

先看看核心要实现Tinkerpop的7个必须数据接口,以及后续的其他类. 我已经把全部代码抽象出来了.注意所有的类基本都是final的(也就是说默认所有方法也是final) ,修饰参数和对象也大量使用final.详见

1. TinkerElement (基类)

实现Element接口, 需要注意的是, 这个元素接口也是Vertex, Edge 接口的基础, 一个元素可以包含多个属性对象.结构如下:

/*属性*/
Object id(); //获得图元素唯一标识-->id
String label(); //获得图的元素标签-->name
boolean removed = false; //元素是否存在的flag -->图常用思路
  
/*方法*/
Object id() //图元素唯一id
String label() //获得图的元素名称
Graph graph() //获得此元素所在的图对象
void remove() //从图中删除此元素
<V> Property<V> property(final String key, final V value) //给定key时,给元素设置属性值

2. TinkerEdge

    /*对边而言,核心数据结构就三个:出入的顶点+属性*/
    Map<String, Property> properties;
    Vertex inVertex;
    Vertex outVertex;

//核心方法
/* A.删除当前边 (5步)
 *1.通过边label,分别获取出顶点的出边和入顶点的入边Set集合,然后移除当前边
 *2.删除当前边的索引
 *3.从图对象中,移除当前边的id
 *4.将当前边的properties置空(null)
 *5.将当前边的remove标记为true
 */
void remove();

/* B. 添加属性 (4步)
 *1.通过key获取旧的属性对象
 *2.根据出key-v生成新的属性对象
 *3.props中put新的属性对象
 *4.更新边属性的索引值 (通过value覆盖oldProperty.value)
 */
Property<V> property(final String key, final V value) 

/* C. 获取当前边关联的顶点
 * 通过方向--> IN/OUT/BOTH分别去获取对应顶点的迭代器
 */
Iterator<Vertex> vertices(final Direction dir)

3. TinkerProperty<V>

/*属性*/
Element ele; //这里放Element是为了通用的表示 --> Vertex/VertexProperty & Edge
String key; 
V value; //源自Property<V>

/*方法*/
//判断element属于Edge,调用它的properties.remove(key),否则调用VertexProperty的
void remove()

4. TinkerVertex

/*属性*/
Map<String,List<VertexProperty>> properties; //关键在于理解这个地方的List<VP>
Map<String,Set<Edge>> outEdges;
Map<String,Set<Edge>> inEdges;
TinkerGraph graph; //以及这里为什么需要一个graph对象

/*方法*/
/*A. 添加属性的方法有多个,选取最完整的.
*1.获取id值,如果空则自增1取下一个(属性id)
*2.通过Cardinality的值处理VertexProperty. 很粗略
*3.判断是否属于图计算模式(默认不属于),进入处理过程
*4.初始化TKVertexProperty对象和properties,获取/初始化List<VP>,然后把新list存起来
*5.更新顶点的K-V索引
*6.附加kvs到VertexProperty (加工)
*/
<V> VertexProperty<V> property(Cardinality cd, String key,V val,Object... objs)
    
/*B. 添加边
*1.获取/生成一个边id
*2.初始化edge对象 (id,inV,outV,label) 
*3.附加kvs到这条边上
*4.在图对象的edges集合中以id-edge添加这个边 (全局注册)
*5.分别给出顶点和入顶点对象在它们自己的出/入边集合中添加进去. (首尾注册)
*/
Edge addEdge(String label,Vertex vertex, Object... kvs)
    
/*C. 删除顶点
*1.获取此顶点入/出方向的所有边,存在一个临时的List<Edge>集合
*2.依次对其中存在的每一条边执行remove() --> 也就是说顶点相关的边都会被删去.
*3. 把properties置空,删除顶点自身的索引
*4. 从图对象全局删除当前顶点id,然后remove标记置为true
*/
 void remove()
    
/*D.获得出入边/出入顶点/属性的迭代器*/
Iterator<X> xs(X... x) //这里X代表上面三个方法的抽象,并非实际

5. TinkerVertexProperty<V>

注意, TinkerGraph中Cardinality包含在TVP内了(默认也是SINGEL). 在JanusGraph和其他实现一般都会单独抽取出来.

/*属性*/
Map<String,Property> properties; //对比一下TinkerProperty这里是Element
TinkerVertex vertex; //特属的顶点属性
String key;
V value;

/*方法*/
/*A.删除当前顶点属性
*1.从vertex的props中获取List<VertexProperty>, 然后从List中删除当前VertexProperty
* 1.1如果删完后,上述的List长度为0了,那么接着把这个List也删掉,然后移除索引
*2.定义一个原子性的bool(默认给true),
*/
void remove()

6. TinkerIndex<T extends Element>

/*属性*/
//这个是最关键的,为什么需要一个复合的map,K-<K-V>到底是什么
Map<String,Map<Object,Set<T>> index = new ConcurrentHashMap<>();
//索引本身的keys是什么,跟上面index.keys有什么区别
Set<String> indexedKeys = new HashSet<>();
Class<T> indexClass;
TinkerGraph  graph;

/*方法(CURD)*/ 
/*1.添加*/
/*1.1 标准put
准备:我们要在图中添加一个索引, 首先要想我们的入参是什么? 附着的元素,K,V?
A.我们先通过入参的K,获取Map<Obejct,Set<T>>,称之为keyMap.(它也是concurrentHashMap)
A2.如果keyMap为空,那么先把它作为V存入index (比如新建一个索引)
B.从keyMap中通过入参的vlaue作为key获得Set<T> objs
B2.如果objs为空,那么把它作为V存入keyMap
C. 把索引附着的element添加到objs中..
*/
void put(String key,Object v,T element)

/*1.2 添加key索引*/
/*
*A.在indexKeys中添加入参的key
*B.通过Vertex.class.isAssignableFrom(indexClass), 这个方法与instanceof相反
也就是判断当前索引类是否属于顶点的子类.
  -Y:从图对象的顶点集合values -->开启一个parallelStream ,意义是?
  -N:从图对象的边集合的values-->开启一个并发流
    1.map获取属性对象数组
    2.筛选数组中第一个属性非空的obj数组
    3.调用1.1的put方法,将obj[0].value作为v传入,element是a[1]? 啥意思...
*/
void createKeyIndex(String key) 
 
/*2.获取*/
List<T> get(String key,Object v)

/*3.删除索引*/
//思路是逆向的添加,可以先试着回忆一下添加.
void remove(String key,Object v, T element)

/*4.删除元素*/
//首先注意删除索引,和删除元素区别,删除元素无关K-V.
void removeElement(T element)

7. TinkerGraph (核心)

TinkerGraph定义了最多的接口,, Janus中类似JanusGraph的实现StandardJanusGraph. 只是区别在于Janus是一个复杂完整的实现, 所以它在Graph的基础上继续做了两层抽象, 而不是直接使用JanusGraph类去实现Graph接口. 本质上和TG是一样的. (理解这个很重要, 不然你不知道为什么Janus中会莫名其妙抽的那么复杂, 到底谁是幕后boss)

• IdManager接口.很简单就是生成id和 id转换

  1. T getNextId(final TinkerGraph g)
  2. T convert(final Object id)

• 枚举类DefaultIdManager唯一实现了IdManager接口.提供以下几种类型ID

  1. LONG (其实这个就是实际使用的..int也是这个基础转换了一下罢了, Janus也有这样的写法)

     //生成long, 或把其它数字类型和字符串值(131111112)尝试转为Long.
     Long getNextId(final TinkerGraph graph){ //currentID是原子Long,初始值-1
       return Stream.generate(graph.currentId::incrementAndGet) //每次原子性的自增1
         .filter(id -> !graph.vertices.containsKey(id) && !graph.edges.containsKey(id))//找出顶点和边中不包含当前id的(判重)
         .findAny().get(); //查找任何一个就返回
     }

  2. INTERER

     //生成int的,其实就是long转了一下
      public Integer getNextId(final TinkerGraph graph) {
                     return Stream.generate(graph.currentId::incrementAndGet)
                       .map(Long::intValue).
                       filter(id -> !graph.vertices.containsKey(id) && !graph.edges.containsKey(id))
                       .findAny().get();
                 }

  3. UUID : java.util.UUID.randomUUID().一句话搞定..

  4. ANY :默认使用Long的策略,只不过其他类型不会转换,直接提示找不到..不知道用途是什么..

---

8. TinkerFactory

这里有几个工具和辅助类, 包括序列化与反序列化的类. 和一张图本身元素的存储类(类似Janus中的SystemSchemaXxx . 只是叫法不同). 之后补充

9. TinkerRegistry

12. TinkerRegistryV2d0

10. TinkerGraphVariables

11. TinkerHelper

这个类其实是个中介, 常见的顶点/边/属性类其实都经常依赖它的方法, 但是其实底层并非自己实现.(它的模板来自Tinkerpop原本的ElementHelper 类). TkHelper有几个核心方法:

1. static Edge addEdge()
   • void addInEdge() //这两个方法其实基本一样,可以整合为addInOrOutEdge()的
   • void addOutEdge()
2. Iterator<TinkerEdge> getEdges()
3. Iterator<TinkerVertex> getVertices()

0x02.源码精选

//注:代码已经阅读完,这里贴的是核心方法, 以及少部分细节供参考,已大量删减.
public final class TinkerProperty<V> implements Property<V> {
    protected final Element element;
    protected final String key;
    protected V value;

    @Override
    public void remove() {
        if (this.element instanceof Edge) {
            ((TinkerEdge) this.element).properties.remove(this.key);
            TinkerHelper.removeIndex((TinkerEdge) this.element, this.key, this.value);
        } else {
            ((TinkerVertexProperty) this.element).properties.remove(this.key);
        }
    }
}

//什么是vertexProperty,这是个比较难理解的地方. 注意对比数据结构和方法
public class TinkerVertexProperty<V> extends TinkerElement implements VertexProperty<V> {
    protected Map<String, Property> properties;
    private final TinkerVertex vertex;
    private final String key;
    private final V value;

    @Override
     //这个方法是没太明白点
    public void remove() {
        if (null != this.vertex.properties && this.vertex.properties.containsKey(this.key)) {
            this.vertex.properties.get(this.key).remove(this);
            if (this.vertex.properties.get(this.key).size() == 0) {
                this.vertex.properties.remove(this.key);
                TinkerHelper.removeIndex(this.vertex, this.key, this.value);
            }
            //关键就在这,为什么要单独定义一个原子的bool.然后去判断是否相同?
            final AtomicBoolean delete = new AtomicBoolean(true);
            this.vertex.properties(this.key).forEachRemaining(property -> {
                if (property.value().equals(this.value))
                    delete.set(false);
            });
            
            if (delete.get()) TinkerHelper.removeIndex(this.vertex, this.key, this.value);
            this.properties = null;
            this.removed = true;
        }
    }

    @Override
    public <U> Iterator<Property<U>> properties(final String... propertyKeys) {
        if (null == this.properties) return Collections.emptyIterator();
        if (propertyKeys.length == 1) {
            final Property<U> property = this.properties.get(propertyKeys[0]);
            return null == property ? Collections.emptyIterator() : IteratorUtils.of(property);
        } else
            return (Iterator) this.properties.entrySet().stream().filter(entry -> ElementHelper.keyExists(entry.getKey(), propertyKeys)).map(entry -> entry.getValue()).collect(Collectors.toList()).iterator();
    }
}

//索引的实现
final class TinkerIndex<T extends Element> {

    protected Map<String, Map<Object, Set<T>>> index = new ConcurrentHashMap<>();
    protected final Class<T> indexClass;
    private final Set<String> indexedKeys = new HashSet<>();
    private final TinkerGraph graph;


    protected void put(final String key, final Object value, final T element) {
        Map<Object, Set<T>> keyMap = this.index.get(key);
        if (keyMap == null) {
            keyMap = new ConcurrentHashMap<>();
            this.index.put(key, keyMap);
        }
        Set<T> objects = keyMap.get(value);
        if (null == objects) {
            objects = new HashSet<>();
            keyMap.put(value, objects);
        }
        objects.add(element);

    }

    public List<T> get(final String key, final Object value) {
        final Map<Object, Set<T>> keyMap = this.index.get(key);
        if (null == keyMap) {
            return Collections.emptyList();
        } else {
            Set<T> set = keyMap.get(value);
            if (null == set)
                return Collections.emptyList();
            else
                return new ArrayList<>(set);
        }
    }


    public void remove(final String key, final Object value, final T element) {
        final Map<Object, Set<T>> keyMap = this.index.get(key);
        if (null != keyMap) {
            Set<T> objects = keyMap.get(value);
            if (null != objects) {
                objects.remove(element);
                if (objects.size() == 0) {
                    keyMap.remove(value);
                }
            }
        }
    }

    public void removeElement(final T element) {
        if (this.indexClass.isAssignableFrom(element.getClass())) {
            for (Map<Object, Set<T>> map : index.values()) {
                for (Set<T> set : map.values()) {
                    set.remove(element);
                }
            }
        }
    }

    public void createKeyIndex(final String key) {
        validate(key);//抽取
        this.indexedKeys.add(key);

        (Vertex.class.isAssignableFrom(this.indexClass) ?
                this.graph.vertices.values().<T>parallelStream() :
                this.graph.edges.values().<T>parallelStream())
                .map(e -> new Object[]{((T) e).property(key), e})
                .filter(a -> ((Property) a[0]).isPresent())
                .forEach(a -> this.put(key, ((Property) a[0]).value(), (T) a[1]));
    }
}

       /**********************图的实现,默认内存中,可以持久化*********************/
public final class TinkerGraph implements Graph {
    static { //每个图实现的最开始先确定遍历的方式, 以及全局的缓存. JanusGraph同
        TraversalStrategies.GlobalCache.registerStrategies(TinkerGraph.class,
    TraversalStrategies.GlobalCache.getStrategies(Graph.class).clone().
    addStrategies(TinkerGraphStepStrategy.instance()));         
    }
    //可以看到,下面是各种图元素的实例合集
    protected AtomicLong currentId = new AtomicLong(-1l);
    protected Map<Object, Vertex> vertices = new ConcurrentHashMap<>();
    protected Map<Object, Edge> edges = new ConcurrentHashMap<>();

    protected TinkerGraphVariables variables = null;
    protected TinkerGraphComputerView graphComputerView = null;
    protected TinkerIndex<TinkerVertex> vertexIndex = null;
    protected TinkerIndex<TinkerEdge> edgeIndex = null;
    //ID管理器的合集
    protected final IdManager<?> vertexIdManager;
    protected final IdManager<?> edgeIdManager;
    protected final IdManager<?> vertexPropertyIdManager;
    protected final VertexProperty.Cardinality defaultVertexPropertyCardinality;

    private final Configuration configuration;
    private final String graphLocation;
    private final String graphFormat;

    /**空的私有构造方法*/
    private TinkerGraph(final Configuration configuration) {
        if (graphLocation != null) loadGraph();
    }


    //////////////核心的CURD的综合实现+调用 //////////////////
    @Override
    public Vertex addVertex(final Object... keyValues) {
        ElementHelper.legalPropertyKeyValueArray(keyValues);
        Object idValue = vertexIdManager.convert
            (ElementHelper.getIdValue(keyValues).orElse(null));
        final String label = 
            ElementHelper.getLabelValue(keyValues).orElse(Vertex.DEFAULT_LABEL);

        if (null != idValue) {
            if (this.vertices.containsKey(idValue))
                throw Exceptions.vertexWithIdAlreadyExists(idValue);
        } else {
            idValue = vertexIdManager.getNextId(this);
        }

        final Vertex vertex = new TinkerVertex(idValue, label, this);
        this.vertices.put(vertex.id(), vertex);
        ElementHelper.attachProperties(vertex, VertexProperty.Cardinality.list, keyValues);
        return vertex;
    }

    @Override
    public <C extends GraphComputer> C compute(final Class<C> graphComputerClass) {
        if (!graphComputerClass.equals(TinkerGraphComputer.class))
            throw 
    Graph.Exceptions.graphDoesNotSupportProvidedGraphComputer(graphComputerClass);
        return (C) new TinkerGraphComputer(this);
    }

    @Override
    public <I extends Io> I io(final Io.Builder<I> builder) {
        return (I) builder.graph(this).onMapper(mapper -> 
            mapper.addRegistry(TinkerIoRegistry.getInstance())).create();
    }

    //全清理
    public void clear() {
        this.vertices.clear();
        this.edges.clear();
        this.variables = null;
        this.currentId.set(-1l);
        this.vertexIndex = null;
        this.edgeIndex = null;
        this.graphComputerView = null;
    }

    @Override
    public void close() { //这里可看到也是可以落到磁盘的
        if (graphLocation != null) saveGraph();
    }

    private void loadGraph() {
        final File f = new File(graphLocation);
        if (f.exists() && f.isFile()) {
            try {
                if (graphFormat.equals("graphml")) {
                    io(IoCore.graphml()).readGraph(graphLocation);
                } else if (graphFormat.equals("graphson")) {
                    io(IoCore.graphson()).readGraph(graphLocation);
                } else if (graphFormat.equals("gryo")) {
                    io(IoCore.gryo()).readGraph(graphLocation);
                } else {
             io(IoCore.createIoBuilder(graphFormat)).readGraph(graphLocation);
                }
            } catch (Exception ex) {
                throw new RuntimeException(String.format("Could not load graph at 
                    %s with %s", graphLocation, graphFormat), ex);
            }
        }
    }

    private void saveGraph() {
        final File f = new File(graphLocation);
        if (f.exists()) {
            f.delete();
        } else {
            final File parent = f.getParentFile();
            //因为相对路径,上一层级文件可能为空
            if (parent != null && !parent.exists()) {
                parent.mkdirs();
            }
        }

        try {
            if (graphFormat.equals("graphml")) {
                io(IoCore.graphml()).writeGraph(graphLocation);
            } else if (graphFormat.equals("graphson")) {
                io(IoCore.graphson()).writeGraph(graphLocation);
            } else if (graphFormat.equals("gryo")) {
                io(IoCore.gryo()).writeGraph(graphLocation);
            } else {
                io(IoCore.createIoBuilder(graphFormat)).writeGraph(graphLocation);
            }
        } catch (Exception ex) {
            throw new RuntimeException(String.format("Could not save graph at %s with %s", graphLocation, graphFormat), ex);
        }
    }

private <T extends Element> Iterator<T> createElementIterator(final Class<T> clazz, 
final Map<Object, T> elements,final IdManager idManager, final Object... ids) {
        final Iterator<T> iterator;
        if (0 == ids.length) {
            iterator = elements.values().iterator();
        } else {
            final List<Object> idList = Arrays.asList(ids);
            validateHomogenousIds(idList);

            /* 如果是Element类型:需要检查每个元素,因为它可能是一个可加工的实例
             *或其他实现, 假设id不需要转换stuff - doesn't seem likely someone would
             *detach a Titan vertex then try to 假设顶点在OrientDB中是可见的
             */
            return clazz.isAssignableFrom(ids[0].getClass()) ?
               IteratorUtils.filter(IteratorUtils.map(idList, id -> 
                elements.get(clazz.cast(id).id())).iterator(), Objects::nonNull)
                : IteratorUtils.filter(IteratorUtils.map(idList, id -> 
            elements.get(idManager.convert(id))).iterator(), Objects::nonNull);
        }
        return TinkerHelper.inComputerMode(this) ?
                (Iterator<T>) (clazz.equals(Vertex.class) ?
                     IteratorUtils.filter((Iterator<Vertex>) iterator, t -> 
                        this.graphComputerView.legalVertex(t)) :
                      IteratorUtils.filter((Iterator<Edge>) iterator, t -> 
                this.graphComputerView.legalEdge(t.outVertex(), t))) :iterator;
 }

///////////// 图索引的CURD方法 ///////////////

    /**
     * 为可见的元素建立索引-->Vertex or Edge 以及 propertyKey.
     * 无论何时一个元素拥有的the specified key mutated, 这个索引将 会更新.
     */
    public <E extends Element> void createIndex(final String key, final Class<E> elementClass) {
        if (Vertex.class.isAssignableFrom(elementClass)) {
            if (null == this.vertexIndex) this.vertexIndex = new TinkerIndex<>(this, TinkerVertex.class);
            this.vertexIndex.createKeyIndex(key);
        } else if (Edge.class.isAssignableFrom(elementClass)) {
            if (null == this.edgeIndex) this.edgeIndex = new TinkerIndex<>(this, TinkerEdge.class);
            this.edgeIndex.createKeyIndex(key);
        } else {
   throw new IllegalArgumentException("Class is not indexable: " + elementClass);
        }
    }


    public <E extends Element> void dropIndex(final String key, final Class<E> elementClass) {
        if (Vertex.class.isAssignableFrom(elementClass)) {
            if (null != this.vertexIndex) this.vertexIndex.dropKeyIndex(key);
        } else if (Edge.class.isAssignableFrom(elementClass)) {
            if (null != this.edgeIndex) this.edgeIndex.dropKeyIndex(key);
        } else {
   throw new IllegalArgumentException("Class is not indexable: " + elementClass);
        }
    }

    public <E extends Element> Set<String> getIndexedKeys(final Class<E> elementClass) {
        if (Vertex.class.isAssignableFrom(elementClass)) {
            return null == this.vertexIndex ? Collections.emptySet() : 
                 this.vertexIndex.getIndexedKeys();
        } else if (Edge.class.isAssignableFrom(elementClass)) {
            return null == this.edgeIndex ? Collections.emptySet() : 
                this.edgeIndex.getIndexedKeys();
        } else {
     throw new IllegalArgumentException("Class is not indexable: " + elementClass);
        }
    }
    
    /*这个就是所谓的ID生成器,核心就两个方法:生成ID, 转换其他类型为ID*/
    public interface IdManager<T> { 
        T getNextId(final TinkerGraph graph);
        T convert(final Object id);
        boolean allow(final Object id);
    }
   }
}

0x03.几个问题

1.VertexProperty和Property的关系

VertexPropert类似于Property，只不过它专门表示Vertex关联的K-V对.但在某种意义上是不同的，因为它还可以表示一个可以拥有属性的Element的对象。

Property很像Java8的Optional类，因为属性不存在（即为空）。 Property的key始终是String，value是object。但每个图db实现的时候, 都会限制特定对象用作value

看了这个可能不太明白, 看看源码设计. Element 设计的是 Vertex,和Edege的基类.它们都继承了Element ,但是Property并没有继承Element,是定义一个Element可以拥有多个Property对象. 但是VertexProperty就比较特别,它继承了Property和Element. 所以如引用所说,VertexProperty可以获得Property… (当然,我并没有搞明白这个区别有什么意义?)

并且public enum Cardinality {SINGEL, LIST, SET} 也是默认定义在VertexProperty接口中的,只不过Janus实现的时候抽取为单独的类了.

2.SystemSchema和SystemVariables的关系

首先我们可以认同,由vertexLabel,EdgeLabel,PropertyKey 这种名字组成的一个系统图(表),类似mysql中的meta信息表是肯定存在的. 只不过实现上可能名字略有差别. (TinkerGraph里是SystemVariables类)

具体来说,只是单纯这个区别么? 这个有待进一步验证, 在JanusGraph中 ,大量存在SystemSchemaXX 的类, 代表系统的内置顶点和表, 以及系统信息, 但是由于继承关系比较复杂, 所以很难快速判断.