vlambda博客
学习文章列表

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) }……}