深入浅出 jackrabbit 之四 索引提交（上）

在上上篇文章中，我们了解了创建索引的一般流程，在上篇文章中，我们也已经明确的知道，一个node的哪些信息是需要加入到索引中去的，也就是jackrabbit是如何来创建document的。那么接下来我们就是要了解jackrabbit是如何把这些document加入到索引文件中去的。

接下来我们需要回顾一下索引概览中的最后几句话：
那么代码的主要逻辑在哪里呢？
在DeleteNode和AddNode类中，还有就是在flush方法中。

这句话充分了说明本文需要描述的内容，那便是
1 如何把document从index删除
2 如何把document放到index中
3 什么是flush

在讨论这3个内容之前我们有必要说明一下executeAndLog方法的作用，从方法名，我们得到的信息是：1执行Node，而log一些数据，那么我们再来看看源代码
executeAndLog(new DeleteNode(transactionId, (UUID) remove.next()));

先进入了executeAndLog方法：

private Action executeAndLog(Action a)
            throws IOException {
        a.execute(this);
        redoLog.append(a);
        // please note that flushing the redo log is only required on
        // commit, but we also want to keep track of new indexes for sure.
        // otherwise it might happen that unused index folders are orphaned
        // after a crash.
        if (a.getType() == Action.TYPE_COMMIT || a.getType() == Action.TYPE_ADD_INDEX) {
            redoLog.flush();
            // also flush indexing queue
            indexingQueue.commit();
        }
        return a;
}

 

在上面这个方法中，我们可以看到以下几个流程：
1 执行DeleteNode的execute方法（是Action的实现类）
2 记录redolog（redolog做啥呀，先留着）
3 刷新redolog到磁盘同时commit indexingqueue（如果是事物提交或者添加index的时候，同时，这个indexingqueue是一个非常怪的设计）

显然这个方法是用来执行Action的，DeleteNode只是Action之一，所以要理解整个体系，我们得先了解了解Action接口到底有多少个实现类：

从图中可以看到Action有很多的实现类，不过ahuaxuan已经在图中作了一些概要的说明以帮助我们理解这些实现类的作用，而且DeleteNode和AddNode类已经出现在我们要分析的范围之内了。

在了解了Action之后，我们需要正视我们在文章开始提出的3个问题了

1 如何把document从index删除

首先，看看DeleteNode里面的逻辑：从上图中我们可以确切的知道DeleteNode就是用来把一个Node从index删除的action。它的主要逻辑都集中在execute方法中，那么下面我们来看看DeleteNode的execute方法（其中包含了ahuaxuan的一些注释和原有的注释，可以帮助我们快速理解它的功能）：

public void execute(MultiIndex index) throws IOException {
//一个deleteNode代表了一个需要被删除的Node，它持有这个//node的uuid
            String uuidString = uuid.toString();
            // check if indexing queue is still working on
            // this node from a previous update
//根据uuid把document从indexingqueue中删除
            Document doc = index.indexingQueue.removeDocument(uuidString);
            if (doc != null) {
                Util.disposeDocument(doc);
            }
            Term idTerm = new Term(FieldNames.UUID, uuidString);
            // if the document cannot be deleted from the volatile index
            // delete it from one of the persistent indexes.
//同时也要从document内存队列或者内存索引中删除，如果存在于volatieindex的document内存队列或者内存索引中，则表示不存在
于persistentindex中，反之亦然。
            int num = index.volatileIndex.removeDocument(idTerm);
            if (num == 0) {
                for (int i = index.indexes.size() - 1; i >= 0; i--) {
//不存在于内存索引中，所以从注册过的persistentindex中删除
                    // only look in registered indexes
                    PersistentIndex idx = (PersistentIndex) index.indexes.get(i);
                    if (index.indexNames.contains(idx.getName())) {
                        num = idx.removeDocument(idTerm);
                        if (num > 0) {
                            return;
                        }
                    }
                }
            }
        }

 

这段方法的主要逻辑是从indexingqueue中删除node对应的document，并从volatileindex对象(或者indexfile)中也删除对应的document，最后从persistentindex类中删除对应的document，所以这里有几件事情值得我们注意：
1． indexingqueue中有可能保存着一个node对应的document，那么这个indexingqueue是做什么用的？
Indexingqueue中的document其实是那些需要的extract的node，因为有些node包含的二级制文件，比如pdf，提炼文本的时候需要点时间，于是，jackrabbit就把提炼的过程给弄成异步的了，提炼完成之后，会把这个document放到indexingqueue中。所以当一个document需要删除的时候，肯定要检查这个异步的队列。

2． Volatileindex是个什么玩意，内存索引，即把索引存放在memory中
但是这里有点蹊跷的地方，需要我们注意一下，我们看看它逻辑：

int removeDocument(Term idTerm) throws IOException {
        Document doc = (Document) pending.remove(idTerm.text());
        int num;
        if (doc != null) {
            Util.disposeDocument(doc);
            // pending document has been removed
            num = 1;
        } else {
            // remove document from index
            num = super.getIndexReader().deleteDocuments(idTerm);
        }
        numDocs -= num;
        return num;
}

 

这个方法在DeleteNode的execute方法中被调用过了，从这个方法来看，这里面有两件事情值得我们关注，一个是volatileindex类中有一个pending队列，放着document，经查，当这个队列超过一定长度的时候，document中的数据就会被写成二进制索引数据放到内存中。
另外一个是如果pending中不存在这个document，那么就会调用indexreader来删除ramdirectory中符合条件的document。
也就是不管怎么做，volatile都是在操作内存，它有一个document队列，当队列长度超过10的时候，这些document就会被转换成index的二进制数据放到ramdirectory。
看到这里，童鞋们应该会问一个问题，就是内存中的index数据怎么写到磁盘中呢？后面我们会看到它有一个copy方法负责把内存中的数据copy到磁盘上。

3． Persistentindex是什么玩意，持久化索引，把索引存放在持久化介质中，目前是local file system。而且看上去PersistentIndex有很多个。不同的persistentindex会使用不同的directory。这种设计是不是告诉我们，这些不同的fsdirectory在某个固定的点是需要合并的呢？带着两个疑问，我们可以继续往下看。
总结成一句话就是，一个document可能存在于多个地方，当删除一个node时，所有的这些地方都需要清扫一遍。

我们来看看第二个大问题：
2 如何把document放到index中
如果DeleteNode一样，AddNode的逻辑也在它的execute方法中，既然如此，我们不妨过去看看：

public void execute(MultiIndex index) throws IOException {
            if (doc == null) {
                try {
//如果doc为null再次创建document，创建document的流程之前已经讲得很清楚了。
                    doc = index.createDocument(new NodeId(uuid));
                } catch (RepositoryException e) {
                    // node does not exist anymore
                    log.debug(e.getMessage());
                }
            }
            if (doc != null) {
                index.volatileIndex.addDocuments(new Document[]{doc});
            }
        }

 

这个方法简单得不能再简单，简单得让人无法置信，呵呵，不要被假象所迷惑。我们来看看
volatileIndex.addDocuments这个方法，因为逻辑都在这个方法中：

void addDocuments(Document[] docs) throws IOException {
        for (int i = 0; i < docs.length; i++) {
            Document old = (Document) pending.put(docs[i].get(FieldNames.UUID), docs[i]);
//这里的pending就是volatileindex中那个document内存队列
            if (old != null) {
                Util.disposeDocument(old);
            }

//如果队列长度超过10（bufferSize），那么执行commitPending，也就是说逻辑又跑到commitPending中去了
            if (pending.size() >= bufferSize) {
                commitPending();
            }
            numDocs++;
        }
        invalidateSharedReader();
    }

 

既然逻辑跑到commitPending中去了，我们就看commitPending，从名字上来看，commitPending就是把pending中的document处理掉：

private void commitPending() throws IOException {
        super.addDocuments((Document[]) pending.values().toArray(
                new Document[pending.size()]));
//从这里可以看出，一旦pending被处理，那么就把pending置空。
        pending.clear();
    }

 

逻辑还不在这个方法中，那么我们继续追踪旅程，来看看这个方法AbstractIndex.addDocuments，在这个方法中，我们终于看到我们想看到的：

void addDocuments(Document[] docs) throws IOException {
        final IndexWriter writer = getIndexWriter();
//一般情况下，我们的docs数组的长度为10，下面就创建10个线程
        DynamicPooledExecutor.Command commands[] =
                new DynamicPooledExecutor.Command[docs.length];

        for (int i = 0; i < docs.length; i++) {
            // check if text extractor completed its work
            final Document doc = getFinishedDocument(docs[i]);
            // create a command for inverting the document
            commands[i] = new DynamicPooledExecutor.Command() {
                public Object call() throws Exception {
                    long time = System.currentTimeMillis();
//每个线程都使用往writer里加入document对象，这个时候，lucene开始解析document，并生产index数据
                    writer.addDocument(doc);
                    return new Long(System.currentTimeMillis() - time);
                }
            };
        }
//并发执行
        DynamicPooledExecutor.Result results[] = EXECUTOR.executeAndWait(commands);
//置空readOnlyReader和sharedReader，为啥置空啊，index数据改了呗
        invalidateSharedReader();
        IOException ex = null;

//检查每个线程的执行情况，有一个出错就抛出异常，其他的异常保存到log中
        for (int i = 0; i < results.length; i++) {
            if (results[i].getException() != null) {
                Throwable cause = results[i].getException().getCause();
                if (ex == null) {
                    // only throw the first exception
                    if (cause instanceof IOException) {
                        ex = (IOException) cause;
                    } else {
                        IOException e = new IOException();
                        e.initCause(cause);
                        ex = e;
                    }
                } else {
                    // all others are logged
                    log.warn("Exception while inverting document", cause);
                }
            } else {
                log.debug("Inverted document in {} ms", results[i].get());
            }
        }
        if (ex != null) {
            throw ex;
        }
    }

 

在看过方法中的注释，大家应该都明白了，在AddNode方法中执行的操作其实就是在内存中生成index数据的操作。而且这个操作是并发执行的。

那么我们内存中的index数据就不需要写到磁盘上了吗，可能吗，要回答这个问题我们回到MultiIndex#update方法：

executeAndLog(new AddNode(transactionId, doc));
                    // commit volatile index if needed
                    flush |= checkVolatileCommit();

 

我们看到这里有一个checkVolatileCommit，就它了，从名字上看，它Y滴就是想把内存中的数据刷到磁盘上，进去看看呗：

private boolean checkVolatileCommit() throws IOException {
        if (volatileIndex.getNumDocuments() >= handler.getMinMergeDocs()) {
            commitVolatileIndex();
            return true;
        }
        return false;
}

 

果然，volatileIndex.getNumDocuments()会返回它处理的document的数量，而handler.getMinMergeDocs()的默认值是100，也就是说当内存中的index数据对应的document超过100的话，就需要做commitVolatileIndex操作，而且返回true，否则就是返回false，这个true或false非常重要，因为它决定着下面的flush操作是否需要操作，而flush就是我们后面要提到的第3个大问题，如果你已经忘记有flush这回事，请看看文章开头提到的3个大问题。那么，到这里理论上来讲内存中的数据现在就需要被刷到磁盘上，ahuaxuan之前也是这么想滴，但是jackrabbit往往出乎人的意料，那么我们一起进去看看（请注意ahuaxuan的注释）：

 

private void commitVolatileIndex() throws IOException {

        // check if volatile index contains documents at all
        if (volatileIndex.getNumDocuments() > 0) {

            long time = System.currentTimeMillis();
            // create index
/*这里创建的是一个Action的实现类CreateIndex，它的作用是创建一个persistentindex对象，很显然，创建的逻辑应该在它的execute方法中，如果你深入的看下去，会发现，这里的传进去的第二参数null，其实是表示创建一个新的persistentindex，而创建一个persistentindex意味着需要一个fsdiretory，创建一个fsdirectory又意味着需要一个目录，于是我们可以看到其实创建index directory的命名方法*/
            CreateIndex create = new CreateIndex(getTransactionId(), null);
            executeAndLog(create);

            // commit volatile index
/*从VolatileCommit这个名字上看来，这个action的主要操作就是把内存里的index数据写到刚创建的 persistentindex对象中，因为需要一个persistentindex的name作为构造参数，恰巧它是新建的这个 persistentindex对象，鉴于方法的重要性，ahuaxuan还是再多罗嗦一下，它的主要功能是把volatile中index的二进制数据拷贝到persistentindex，其实就是数据从ramdirectory向fsdirectory转移的过程 */

            executeAndLog(new VolatileCommit(getTransactionId(), create.getIndexName()));

            // add new index
/*这个AddIndex类也是非常重要的类，它存在的目的是什么呢？*/
            AddIndex add = new AddIndex(getTransactionId(), create.getIndexName());
            executeAndLog(add);

            // create new volatile index
/*volatileindex 的使命到这里就结束了，下面就重新创建一个volatileindex吧，放弃过去，重新开始 */
            resetVolatileIndex();

            time = System.currentTimeMillis() - time;
            log.debug("Committed in-memory index in " + time + "ms.");
        }
    }

 

从代码和ahuaxuan的注释上看，要把内存中的index数据刷到磁盘上还真不是一件简单的事情。不过从这个过程中，我们得到一个重要的信息，最开始的时候，persistentindex中只包含了100个document的index数据。不过聪明的你一定已经看出点端倪，冥冥之中，好像在告诉我们，所有的这些persistentindex其实都缺少一个操作，它们需要合并index文件，这个操作和AddIndex息息相关，我们貌似对AddIndex的左右已经看出一点端倪了。
要解开这个谜团，非得到AddIndex中看看不可：

public void execute(MultiIndex index) throws IOException {
            PersistentIndex idx = index.getOrCreateIndex(indexName);
/*index.indexNames是个什么东西？其实就是一个列表，任何一个新的persistentindex都需要被注册到这个列表中，它的作用很多，比如注册之后，只需要通过persistentindex类的名字就是可以取到对应的persistentindex，在上面的方法中并没有把新建的persistentindex注册到indexNames中去，所以这个方法先判断有没有注册，没有注册那么就注册进去，接着把这个 indexName传给了merger类，从名字上来看，这个merger的功能应该就是合并persistentindex的数据了*/
            if (!index.indexNames.contains(indexName)) {
                index.indexNames.addName(indexName);
                // now that the index is in the active list let the merger know about it
                index.merger.indexAdded(indexName, idx.getNumDocuments());
            }
        }

 

那么既然知道了有这么一个merger类，那不看看他做了点什么也过意不去啊。
看看它的类注释：Merges indexes in a separate deamon thread.
原来是一个deamon线程啊，那它一定是不停的在后台执行merge操作，而且理论上来讲，也应该有一个临界值，超过多少document被写到persistentindex的时候就执行merger操作。这里面的逻辑应该有很多值得我们学习的地方，抱着这样的想法，ahuaxuan决定把它的研究放到后面。

接下来，让我们说说flush这个操作，这是索引提交的第三部重要流程
  To be continue

 

评论

2 楼 ahuaxuan 2011-12-14  

你好，我分析的是jackrabbit1.7里的代码

1 楼 27g 2011-12-13  

请问一下，您这里分析的代码是lucene的源码么？是的话是哪个版本的？