深入浅出 jackrabbit 六 文本提取（上)

用lucene作过索引的同学一定对文本提取不会陌生，但凡有binary需要作索引的地方，那就离不开文本的提取，而jackrabbit是可以存储二进制数据的，而这些二进制数据中包含的文本也是需要被搜索到的，所以jackrabbit里建索引的过程中自然离不开文本提取。

那么看到本文的标题，有经验的同学一定会问，前面都已经讲过索引的提交了，为什么到现在才讲文本提取呢。要回答这个问题，我们必须来回顾一下jackrabbit中建索引的流程（简单的划分为4个步骤，如下）：

1创建document-----2加入到volatileindex的pending缓存中---------3生成index数据到内存-------4将内存中的二进制数据刷到文件系统。
那么哪里需要提取文本呢，很显然是在创建document的时候。那这个话题是不是应该放到创建document一文中阐述比较好呢，非也，这个问题之所以放到索引提交后面阐述是有原因的。

一般来说，文本提取的耗时长度和文本的大小有直接关系，当文本的大小超过一定的量之后，提取的速度会比较慢。那么当前的WorkThread就会一直耗在这块地方，也就是上述流程中的第一步，这样WorkThread一定要等到文本提取之后才能返回，不知道有多少用户有这个耐性等待下去呢。而且这里还有另外一个问题，那就是WorkThread如果因为这种操作而阻塞住，那么它回线程池的时间就会延后，新的请求过来因为拿到不到线程池中的WorkThread而不得不等待，一直到池中有可用线程为止，或者选择timeout返回，那么如果因为WorkThread都耗在文本提取这个地方，那么就太冤了，因为一般来说大文本做索引的实时性要求并不是很高，所以因为同步文本提取这个原因导致系统的并发量急剧下降，用户就会开始抱怨，你们这个破玩意怎么会这么慢？

为了解决这个问题，在jackrabbit中引入了异步处理文本提取任务的功能（它必须是异步的吗？当然不，jackrabbit也提供了同步的选择，但是本文主要阐述异步提取的功能，所以在下面的文章中我们假设我们配置异步提取的功能），说到这里，我们不得不再说说他是怎么个异步法的呢？它是异步提取，然后由处理提取的线程负责把这个document重新加入到index中呢，还是提取的线程只负责提取，提取完成之后再把提取完成的document交给专门的线程去add到document中呢。

后者，当然是后者，提取的线程只负责提取，有专门的线程来负责更新document到index中。如果说观察者模式是设计模式中的皇后，那么ahuaxuan可以说，生产消费模型是编程模型中的皇后，它无处不在（您要是天天写crud的话，ahuaxuan很难保证它无处不在）。

在前面的文章中，我们一直在提MultiIndex#indexingqueue，而且一直没有详细解释indexingqueue的功能。其实它就是在异步文本提取中充当着消费队列的职责，提取线程（生产者）再文本提取完成之后，会把document加入到indexingqueue中。那么消费者线程就负责定时的从indexingqueue中取出一件准备好的document，并执行multiindex#update方法。

我们先来看看提取文本是怎么做的。再次回到NodeIndexer类的addBinary方法，我相信大家对这个方法还有印象，他负责创建一个名字叫jcr:data的field。他的值就是来自于二进制文件，比如pdf，doc等等：

protected void addBinaryValue(Document doc,
                                  String fieldName,
                                  Object internalValue) {
       ………………………………………………..
       
                Reader reader;
                if (extractor instanceof PooledTextExtractor) {
//从类名上我们也可以看出点端倪，这里貌似有个线程池。
                    reader = ((PooledTextExtractor)extractor).extractText(stream, type, encoding, node.getNodeId().getUUID().toString());
                } else {
//忽视这个else，假设只进入if，                    reader = extractor.extractText(stream, type, encoding);
                }
//这个createFulltextField方法非常之相当重要。后面会讲到
                doc.add(createFulltextField(reader));
            ……………………………………………………………..
    }

 

现在我们要做的就是跟到PooledTextExtractor# extractText方法中去，其他能做的也不多：

public Reader extractText(InputStream stream,
                              String type,
                              String encoding,
                              String jobId) throws IOException {
        if (null == jobId || "".equals(jobId)) {
//结合上面的方法，我们得知，jobid的值是一个node的id
            return extractText(stream, type, encoding);
        } else {
/*在这里，创建了一个TextExtractorJob，它是Runnable的一个实现类*/
            TextExtractorJob job = new TextExtractorJob(extractor, stream, type, encoding, jobId);
            return new TextExtractorReader(job, executor, timout);
        }
    }

 

从上面的方法中，我们可以看到一个简单的逻辑，创建一个thread，然后放到池中执行，并返回结果。
先看看创建线程的流程：

public TextExtractorJob(final TextExtractor extractor,
                            final InputStream stream,
                            final String type,
                            final String encoding,
                            final String jobId) {
        this.type = type;
        this.jobId = jobId;
        this.cmd = setter(new Callable() {
            public Object call() throws Exception {
                Reader r = extractor.extractText(stream, type, encoding);
                if (r != null) {
                    if (discarded) {
                        r.close();
                        r = null;
                    } else if (timedOut) {
                        // spool a temp file to save memory
/*这里的逻辑非常重要，这里的逻辑表明，一旦方法执行到这里，表示上面的extractor.extractText 已经超过了我们预设的时间，所以为了避免占用过多的内存，需要把内存中的数据拷贝到磁盘上，以减少内存的消耗。*/
                        r = getSwappedOutReader(r);
                    }
                }
                return r;
            }
        });
    }

 

在上面的这个方法中，我们看到在TextExtractorJob这个线程类内部还有一个线程类，就是cmd，而且这个线程类有一个奇怪的地方，cmd这个线程的run方法中其实是执行一个Callable接口的实现类（这个逻辑写在了setter（Callable cc）中）。cmd的逻辑其实就是执行真正的文本提取操作。而TextExtractorJob的run方法其实就是执行cmd的run方法。也就是说cmd作为一个线程类实现压根没有被真正的作为一个线程启动过，所以cmd定义为runnable只是给人多一份疑惑，其他作用没有看出来。

创建完TextExtractorJob之后，就要用它来创建TextExtractorReader。我们看看TextExtractorReader的构造方法：

TextExtractorReader(TextExtractorJob job, Executor executor, long timeout) {
        this.job = job;
        this.executor = executor;
        this.timeout = timeout;
}

 

很简单，什么都没有，也就是说在这个操作的过程中，并没有触发文本提取的操作，那么这个操作是什么时候做的呢？这就需要我们回到addBinary方法，看看它是怎么用这个TextExtractorReader类的：
doc.add(createFulltextField(reader));
原来是又调用了createFulltextField方法，那么我们得进去看一下，说不定触发文本提取的操作就在这里哦：

protected Fieldable createFulltextField(Reader value) {
        if (supportHighlighting) {
            return new LazyTextExtractorField(FieldNames.FULLTEXT, value, true, true);
        } else {
            return new LazyTextExtractorField(FieldNames.FULLTEXT, value, false, false);
        }
    }

 

额的神啊，啥都没有，就创建了一个LazyTextExtractorField类（还记得supportHighlighting参数吗，不记得的话再回头看看前面的文章），也就是说我们不得不再去看看LazyTextExtractorField这个类，但是我们有种感觉，快要看到我们想看到的东西了，这个继承实现了lucene中的AbstractField，结合上面的doc.add方法，我们可以想象得到，真正触发提取线程工作的应该是lucene，lucene一定会调用LazyTextExtractorField类中的某个方法，以得到值，而这个时候，如果有文本提取的任务，那么应该会触发它，然后改方法先返回空字符串，把残缺版的document先加入索引中，等待提取完成后，再由前面提到的消费者把残缺版的document数据从索引中删除，再把完整版的document加入到索引中。
上面说到的索引时触发文本提取操作只是一种触发方式，通过阅读源代码，我们还可以发现第二种触发方式，那就是在AbstractIndex的addDocument方法中，会判断，如果提取超过100ms，就把document拷贝一份，把copy的document加入到index中，然后把原始的document加入到indexingQueue队列中。这个时候，提取线程继续为这个原始的document执行提取操作。
那么我们先来看第一种触发，lucene触发，下面这个方法是LazyTextExtractorField中的方法，这个方法被lucene调用，用以分词, ahuaxuan已经在关键的代码中加入了注释：

public String stringValue() {
        if (extract == null) {
            StringBuffer textExtract = new StringBuffer();
            char[] buffer = new char[1024];
            int len;
            try {
    /*这里的读取操作有一定的迷惑性，其实它是调用了，考虑到这里的reader是TextExtractorReader，所以耗无疑问，这里的方法一定是调用了TextExtractorReader#read */               
while ((len = reader.read(buffer)) > -1) {
                    textExtract.append(buffer, 0, len);
                }
            } catch (IOException e) {
                log.warn("Exception reading value for field: "
                        + e.getMessage());
                log.debug("Dump:", e);
            } finally {
                try {
                    reader.close();
                } catch (IOException e) {
                    // ignore
                }
            }
            extract = textExtract.toString();
        }
        return extract;
    }

 

方法很简单，就是调用reader的读取，只是它调用的是Reader类的read（）方法：

public int read(char cbuf[]) throws IOException {
    return read(cbuf, 0, cbuf.length);
    }
而其实目标的方法是子类的read(cbuf, 0, cbuf.length)；也就是说应该是TextExtractorReader# read(cbuf, 0, cbuf.length),那么我们来看一下这个方法TextExtractorReader# read：这个read方法才是真正为LazyTextExtractorField#stringValue返回数据的地方
    public int read(char cbuf[], int off, int len) throws IOException {
/*事实上，这里的extractedText应该是不会等于空的，因为如果是异步提取，那么在lucene调用read方法之前，extract线程早就把extractedText提取出来，或者将其置为new StringReader(“”)了,除非一个document在isExtractorFinished返回false的时候，也被加入了 indexwriter。在正常流程的debug过程中，ahuaxuan并没有发现程序执行到这个if块里面*/

        if (extractedText == null) {
            // no reader present
            // check if job is started already
            if (jobStarted) {
                // wait until available
/*如果任务已经开始，那么等待读取文本，那么如果任务开始，这个开始是谁开始的呢？其实就是后面要讲到的第二种触发方式,这里是无限时等待，太恐怖了，所以这个代码段是存在危险的,实际上，如果我们如果在每个单独的Extractor类中设置专门的超时时间，那么这里就没有问题了。*/
                extractedText = job.getReader(Long.MAX_VALUE);
            } else {
                // execute with current thread
/*如果提取任务还没有被触发，则把该线程实例放到DIRECT_EXECUTOR这个线程池中运行,注意，其实这个类很诡异，它并不是一个线程池，它拿到线程对象之后会调用run方法，而不是start，这也意味着这个操作是非异步的，这里不会有多线程的问题 */
                try {
                    DIRECT_EXECUTOR.execute(job);
                } catch (InterruptedException e) {
                    // current thread is in interrupted state
                    // -> ignore (job will not return a reader, which is fine)
                }
/*同步执行，立即取结果，取不到就返回null*/
                extractedText = job.getReader(0);
            }

            if (extractedText == null) {
                // exception occurred
                extractedText = new StringReader("");
            }
        }
        return extractedText.read(cbuf, off, len);
    }

 

   
需要注意的是getSwappedOutReader这个方法，该方法用临时文件的方式来节约内存。但不是说什么地方都可以使用这种方法的，因为提取之后的问题可能在后面才能被用到，所以暂时存放在临时文件中是可行的，这种方式在下载工具中非常常见，一般下载工具下载文件时会创建临时文件也是同样的道理。
接着，我们再来看看第二中方式触发（事实上，最开始触发这个方法的地方才是真正的第二种触发，它在什么地方呢，请参考AbstractIndex#getFinishedDocument(xx)，再文本提取的第二篇文章中也有比较详细的说明）：

public boolean isExtractorFinished() {
        if (!jobStarted) {
            try {
/*如果任务还没有开始，那么则把job放到线程池中执行，并且将jobStarted设置为true， */
                executor.execute(job);
                jobStarted = true;
            } catch (InterruptedException e) {
                // this thread is in interrupted state
                return false;
            }
/*然后开始提取的操作，同时，提取的时候有一个超时时间，代表超过多长时间就直接返回，默认时间是100毫秒，在repository.xml中的 SearchIndex节点中可以配置，一旦超时之后，reader就会被写入到临时文件中，下次读取就从临时文件中读取*/
            extractedText = job.getReader(timeout);
        } else {
            // job is already running, check for immediate result
/*如果任务已经在上次检查中被触发过了，那么就直接获取提取之后的结果，如果还没有提取完，0表示没有提取完则抛出TimeoutException，但是getReader会捕获这个异常，并返回一个null*/
            extractedText = job.getReader(0);
        }
/*如果extractedText并且提取的时候出现异常，超时等，那么就置其为空*/
        if (extractedText == null && job.getException() != null) {
            // exception occurred
            extractedText = new StringReader("");
        }

        return extractedText != null;
    }

 

由此可见，这里的逻辑是当地一次提取的时候，等待100毫秒之后返回，如果100毫秒之内搞定问题，那事情很顺利，如果没有搞定呢，那就等下次，下次再进入这个方法，如果还没有搞定，那么reader就是new StringReader("")了，这样二进制就没有能被成功提取出来。
这个方法是谁来调用的呢，还是那个indexingqueue的消费者。在下文中，让我们来分析一下它。

   
好了说到这里，大家应该对jackrabbit中异步文本提取的功能算是比较了解了，之前ahuaxuan说过，jackrabbit也是支持同步文本提取的，这个就比较简单了，相信大家都可以想象得出来，在一个hashmap中存在了一组extractor，比mswordextractor，pdfextractor，那么当需要提取文本时，只要把inputstream和filetype交给处理器，那么，处理器就会为该filetype选择合适的extractor并执行extractText方法，这些逻辑都在CompositeTextExtractor.java类中，根据filetype委派给对应的extractor，其代码也只有90行，大家可以自行查看。

    总结，在本文中，ahuaxuan主要描述了jackrabbit文本提取的两种方式，同步和异步，而且在异步的方式中，触发提取操作的有两个点，一个是消费者检查哪些document已经提取完成的时候，还有一个是lucene调用field的stringValue时候，即从document生成二进制index数据的时候。这样通过阅读源代码，我们详细的知道了jackrabbit中提取文件的逻辑。不过这个过程并不是无可挑剔的，比如存在二进制文件过大，那么提取文本过多，占用过多内存，而且占用很长的cpu时间，如何修改才能使之满足我们的需求呢。

从本文中我们得到如下信息，文本提取的主要逻辑是什么，文本提取的触发点是什么。当然我们也得到了一个疑问，在文件提取的异步模型中，谁是生产者，谁是消费者，具体的流程是怎么样的。

    在下文中，ahuaxuan将把本文提出的细节和整个流程串起来，形成一幅完整的流程。
To be continue