深度解析 Lucene 轻量级全文索引实现原理

ThreadState getAndLock(Thread requestingThread, DocumentsWriter documentsWriter) {  ThreadState threadState = null;  synchronized (this) {    if (freeList.isEmpty()) {      // 如果不存在已创建的空闲ThreadState，则新创建一个      return newThreadState();    } else {      // freeList后进先出，仅使用有限的ThreadState操作索引      threadState = freeList.remove(freeList.size()-1);
      // 优先使用已经初始化过DocumentWriterPerThread的ThreadState，并将其与当前      // ThreadState换位，将其移到队尾优先使用      if (threadState.dwpt == null) {        for(int i=0;i<freeList.size();i++) {          ThreadState ts = freeList.get(i);          if (ts.dwpt != null) {            freeList.set(i, threadState);            threadState = ts;            break;          }        }      }    }  }  threadState.lock();    return threadState;}

// 构建倒排表
if (fieldType.indexOptions() != IndexOptions.NONE) {    fp = getOrAddField(fieldName, fieldType, true);    boolean first = fp.fieldGen != fieldGen;    // field具体的索引、分词操作    fp.invert(field, first);
    if (first) {      fields[fieldCount++] = fp;      fp.fieldGen = fieldGen;    }} else {  verifyUnIndexedFieldType(fieldName, fieldType);}
// 存储该field的storeFieldif (fieldType.stored()) {  if (fp == null) {    fp = getOrAddField(fieldName, fieldType, false);  }  if (fieldType.stored()) {    String value = field.stringValue();    if (value != null && value.length() > IndexWriter.MAX_STORED_STRING_LENGTH) {      throw new IllegalArgumentException("stored field \"" + field.name() + "\" is too large (" + value.length() + " characters) to store");    }    try {      storedFieldsConsumer.writeField(fp.fieldInfo, field);    } catch (Throwable th) {      throw AbortingException.wrap(th);    }  }}
// 建立DocValue（通过文档查询文档下包含了哪些词）DocValuesType dvType = fieldType.docValuesType();if (dvType == null) {  throw new NullPointerException("docValuesType must not be null (field: \"" + fieldName + "\")");}if (dvType != DocValuesType.NONE) {  if (fp == null) {    fp = getOrAddField(fieldName, fieldType, false);  }  indexDocValue(fp, dvType, field);}if (fieldType.pointDimensionCount() != 0) {  if (fp == null) {    fp = getOrAddField(fieldName, fieldType, false);  }  indexPoint(fp, field);}

// invert：对Field进行分词处理首先需要将Field转化为TokenStreamtry (TokenStream stream = tokenStream = field.tokenStream(docState.analyzer, tokenStream))// TokenStream在不同分词器下实现不同，根据不同分词器返回相应的TokenStreamif (tokenStream != null) {  return tokenStream;} else if (readerValue() != null) {  return analyzer.tokenStream(name(), readerValue());} else if (stringValue() != null) {  return analyzer.tokenStream(name(), stringValue());}
public final TokenStream tokenStream(final String fieldName, final Reader reader) {  // 通过复用策略，如果TokenStreamComponents中已经存在Component则复用。  TokenStreamComponents components = reuseStrategy.getReusableComponents(this, fieldName);  final Reader r = initReader(fieldName, reader);  // 如果Component不存在，则根据分词器创建对应的Components。  if (components == null) {    components = createComponents(fieldName);    reuseStrategy.setReusableComponents(this, fieldName, components);  }  // 将java.io.Reader输入流传入Component中。  components.setReader(r);  return components.getTokenStream();}

// 标准分词器创建Component过程，涵盖了标准分词处理器、Term转化小写、常用词过滤三个功能protected TokenStreamComponents createComponents(final String fieldName) {  final StandardTokenizer src = new StandardTokenizer();  src.setMaxTokenLength(maxTokenLength);  TokenStream tok = new StandardFilter(src);  tok = new LowerCaseFilter(tok);  tok = new StopFilter(tok, stopwords);  return new TokenStreamComponents(src, tok) {    @Override    protected void setReader(final Reader reader) {      src.setMaxTokenLength(StandardAnalyzer.this.maxTokenLength);      super.setReader(reader);    }  };}

// 以LowerCaseFilter为例，通过其下的increamentToken将Token中的字符转化为小写public final boolean incrementToken() throws IOException {  if (input.incrementToken()) {    CharacterUtils.toLowerCase(termAtt.buffer(), 0, termAtt.length());    return true;  } else    return false;}

  try (TokenStream stream = tokenStream = field.tokenStream(docState.analyzer, tokenStream)) {    // reset TokenStream    stream.reset();    invertState.setAttributeSource(stream);    termsHashPerField.start(field, first);    // 分析并获取Token属性    while (stream.incrementToken()) {      ……      try {        // 构建倒排表        termsHashPerField.add();      } catch (MaxBytesLengthExceededException e) {        ……      } catch (Throwable th) {        throw AbortingException.wrap(th);      }    }    ……}

synchronized long deleteTerms(final Term... terms) throws IOException {  // TODO why is this synchronized?  final DocumentsWriterDeleteQueue deleteQueue = this.deleteQueue;  // 文档删除操作是将删除的词信息添加到删除队列中，根据flush策略进行删除  long seqNo = deleteQueue.addDelete(terms);  flushControl.doOnDelete();  lastSeqNo = Math.max(lastSeqNo, seqNo);  if (applyAllDeletes(deleteQueue)) {    seqNo = -seqNo;  }  return seqNo;}

private boolean applyAllDeletes(DocumentsWriterDeleteQueue deleteQueue) throws IOException {  // 判断是否满足删除条件 --> onDelete  if (flushControl.getAndResetApplyAllDeletes()) {    if (deleteQueue != null) {      ticketQueue.addDeletes(deleteQueue);    }    // 指定执行删除操作的event    putEvent(ApplyDeletesEvent.INSTANCE); // apply deletes event forces a purge    return true;  }  return false;}

public void onDelete(DocumentsWriterFlushControl control, ThreadState state) {  // 判断并设置是否满足删除条件  if ((flushOnRAM() && control.getDeleteBytesUsed() > 1024*1024*indexWriterConfig.getRAMBufferSizeMB())) {    control.setApplyAllDeletes();    if (infoStream.isEnabled("FP")) {      infoStream.message("FP", "force apply deletes bytesUsed=" + control.getDeleteBytesUsed() + " vs ramBufferMB=" + indexWriterConfig.getRAMBufferSizeMB());    }  }}

@Overridepublic void onInsert(DocumentsWriterFlushControl control, ThreadState state) {  // 根据文档数进行Flush  if (flushOnDocCount()      && state.dwpt.getNumDocsInRAM() >= indexWriterConfig          .getMaxBufferedDocs()) {    // Flush this state by num docs    control.setFlushPending(state);  // 根据内存使用量进行Flush  } else if (flushOnRAM()) {// flush by RAM    final long limit = (long) (indexWriterConfig.getRAMBufferSizeMB() * 1024.d * 1024.d);    final long totalRam = control.activeBytes() + control.getDeleteBytesUsed();    if (totalRam >= limit) {      if (infoStream.isEnabled("FP")) {        infoStream.message("FP", "trigger flush: activeBytes=" + control.activeBytes() + " deleteBytes=" + control.getDeleteBytesUsed() + " vs limit=" + limit);      }      markLargestWriterPending(control, state, totalRam);    }  }}

private synchronized boolean updatePendingMerges(MergePolicy mergePolicy, MergeTrigger trigger, int maxNumSegments)  throws IOException {
  final MergePolicy.MergeSpecification spec;  // 查询需要合并的段  if (maxNumSegments != UNBOUNDED_MAX_MERGE_SEGMENTS) {    assert trigger == MergeTrigger.EXPLICIT || trigger == MergeTrigger.MERGE_FINISHED :    "Expected EXPLICT or MERGE_FINISHED as trigger even with maxNumSegments set but was: " + trigger.name();
    spec = mergePolicy.findForcedMerges(segmentInfos, maxNumSegments, Collections.unmodifiableMap(segmentsToMerge), this);    newMergesFound = spec != null;    if (newMergesFound) {      final int numMerges = spec.merges.size();      for(int i=0;i<numMerges;i++) {        final MergePolicy.OneMerge merge = spec.merges.get(i);        merge.maxNumSegments = maxNumSegments;      }    }  } else {    spec = mergePolicy.findMerges(trigger, segmentInfos, this);  }  // 注册所有需要合并的段  newMergesFound = spec != null;  if (newMergesFound) {    final int numMerges = spec.merges.size();    for(int i=0;i<numMerges;i++) {      registerMerge(spec.merges.get(i));    }  }  return newMergesFound;}

@Overridepublic synchronized void merge(IndexWriter writer, MergeTrigger trigger, boolean newMergesFound) throws IOException {  ……  while (true) {    ……    // 取出注册的后选段    OneMerge merge = writer.getNextMerge();    boolean success = false;    try {      // 构建用于合并的线程MergeThread       final MergeThread newMergeThread = getMergeThread(writer, merge);      mergeThreads.add(newMergeThread);
      updateIOThrottle(newMergeThread.merge, newMergeThread.rateLimiter);
      if (verbose()) {        message("    launch new thread [" + newMergeThread.getName() + "]");      }      // 启用线程      newMergeThread.start();      updateMergeThreads();
      success = true;    } finally {      if (!success) {        writer.mergeFinish(merge);      }    }  }}

public void merge(MergePolicy.OneMerge merge) throws IOException {  ……      try {        // 用于处理merge前缓存任务及新段相关信息生成        mergeInit(merge);        // 执行段之间的merge操作        mergeMiddle(merge, mergePolicy);        mergeSuccess(merge);        success = true;      } catch (Throwable t) {        handleMergeException(t, merge);      } finally {        // merge完成后的收尾工作        mergeFinish(merge)      }……}