深度解析 Lucene 輕量級(jí)全文索引實(shí)現(xiàn)原理

ThreadState getAndLock(Thread requestingThread, DocumentsWriter documentsWriter) {  ThreadState threadState = null;  synchronized (this) {    if (freeList.isEmpty()) {      // 如果不存在已創(chuàng)建的空閑ThreadState，則新創(chuàng)建一個(gè)      return newThreadState();    } else {      // freeList后進(jìn)先出，僅使用有限的ThreadState操作索引      threadState = freeList.remove(freeList.size()-1);
      // 優(yōu)先使用已經(jīng)初始化過(guò)DocumentWriterPerThread的ThreadState，并將其與當(dāng)前      // ThreadState換位，將其移到隊(duì)尾優(yōu)先使用      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;}

// 構(gòu)建倒排表
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);}
// 存儲(chǔ)該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（通過(guò)文檔查詢文檔下包含了哪些詞）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：對(duì)Field進(jìn)行分詞處理首先需要將Field轉(zhuǎn)化為TokenStreamtry (TokenStream stream = tokenStream = field.tokenStream(docState.analyzer, tokenStream))// TokenStream在不同分詞器下實(shí)現(xiàn)不同，根據(jù)不同分詞器返回相應(yīng)的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) {  // 通過(guò)復(fù)用策略，如果TokenStreamComponents中已經(jīng)存在Component則復(fù)用。  TokenStreamComponents components = reuseStrategy.getReusableComponents(this, fieldName);  final Reader r = initReader(fieldName, reader);  // 如果Component不存在，則根據(jù)分詞器創(chuàng)建對(duì)應(yīng)的Components。  if (components == null) {    components = createComponents(fieldName);    reuseStrategy.setReusableComponents(this, fieldName, components);  }  // 將java.io.Reader輸入流傳入Component中。  components.setReader(r);  return components.getTokenStream();}

// 標(biāo)準(zhǔn)分詞器創(chuàng)建Component過(guò)程，涵蓋了標(biāo)準(zhǔn)分詞處理器、Term轉(zhuǎn)化小寫、常用詞過(guò)濾三個(gè)功能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為例，通過(guò)其下的increamentToken將Token中的字符轉(zhuǎn)化為小寫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 {        // 構(gòu)建倒排表        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;  // 文檔刪除操作是將刪除的詞信息添加到刪除隊(duì)列中，根據(jù)flush策略進(jìn)行刪除  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);    }    // 指定執(zhí)行刪除操作的event    putEvent(ApplyDeletesEvent.INSTANCE); // apply deletes event forces a purge    return true;  }  return false;}

public void onDelete(DocumentsWriterFlushControl control, ThreadState state) {  // 判斷并設(shè)置是否滿足刪除條件  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) {  // 根據(jù)文檔數(shù)進(jìn)行Flush  if (flushOnDocCount()      && state.dwpt.getNumDocsInRAM() >= indexWriterConfig          .getMaxBufferedDocs()) {    // Flush this state by num docs    control.setFlushPending(state);  // 根據(jù)內(nèi)存使用量進(jìn)行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);  }  // 注冊(cè)所有需要合并的段  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) {    ……    // 取出注冊(cè)的后選段    OneMerge merge = writer.getNextMerge();    boolean success = false;    try {      // 構(gòu)建用于合并的線程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前緩存任務(wù)及新段相關(guān)信息生成        mergeInit(merge);        // 執(zhí)行段之間的merge操作        mergeMiddle(merge, mergePolicy);        mergeSuccess(merge);        success = true;      } catch (Throwable t) {        handleMergeException(t, merge);      } finally {        // merge完成后的收尾工作        mergeFinish(merge)      }……}