lucene添加文档源码解析(第一篇)

        IndexWriterConfig config = new IndexWriterConfig(new WhitespaceAnalyzer());
        config.setUseCompoundFile(false);
        config.setMaxBufferedDocs(2);
        IndexWriter writer = new IndexWriter(dir, config);
        //
        FieldType type = new FieldType();
        type.setStored(true);
        type.setTokenized(true);
        type.setIndexOptions(IndexOptions.DOCS_AND_FREQS_AND_POSITIONS_AND_OFFSETS);
        type.setStoreTermVectors(true);
        type.setStoreTermVectorPositions(true);
        type.setStoreTermVectorOffsets(true);
        type.freeze();
        //
        Document doc = new Document();
        doc.add(new Field("content", "one", type));
        writer.addDocument(doc);

  private boolean preUpdate() throws IOException, AbortingException {
    ensureOpen();
    boolean hasEvents = false;
    // 判断是否有需要待flush的DWPT
    if (flushControl.anyStalledThreads() || (flushControl.numQueuedFlushes() > 0 && config.checkPendingFlushOnUpdate)) {
      // Help out flushing any queued DWPTs so we can un-stall:
      do {
        // Try pick up pending threads here if possible
        DocumentsWriterPerThread flushingDWPT;
        // 取出每一个DWPT进行flush
        while ((flushingDWPT = flushControl.nextPendingFlush()) != null) {
          // Don't push the delete here since the update could fail!
          // 对该flushingDWPT进行flush
          hasEvents |= doFlush(flushingDWPT);
        }
        
        flushControl.waitIfStalled(); // block if stalled
      } while (flushControl.numQueuedFlushes() != 0); // still queued DWPTs try help flushing
    }
    return hasEvents;
  }

    /*
     * we block indexing threads if net byte grows due to slow flushes
     * yet, for small ram buffers and large documents we can easily
     * reach the limit without any ongoing flushes. we need to ensure
     * that we don't stall/block if an ongoing or pending flush can
     * not free up enough memory to release the stall lock.
     */
     stall = (activeBytes + flushBytes) > limit && activeBytes < limit && !closed;

/** 
  if an indexing thread should check for pending flushes on update 
  in order to help out on a full flush
  */
protected volatile boolean checkPendingFlushOnUpdate = true;

  DocumentsWriterPerThread nextPendingFlush() {
    int numPending;
    boolean fullFlush;
    synchronized (this) {
      final DocumentsWriterPerThread poll;
      // 取出添加到flushQueue中的DWPT
      if ((poll = flushQueue.poll()) != null) {
        // 更新线程停滞标志
        updateStallState();
        return poll;
      }
      fullFlush = this.fullFlush;
      numPending = this.numPending;
    }
    // 如果标记为flushPending状态的ThreadState的个数大于0，并且当前没有处于full flush状态(主动执行flush)
    if (numPending > 0 && !fullFlush) { // don't check if we are doing a full flush
      // 处理每一个ThreadState,如果该ThreadState被标记为flushPending状态，尝试取出该ThreadState
      // 关联的的DWPT，返回该DWPT进行flush
      final int limit = perThreadPool.getActiveThreadStateCount();
      for (int i = 0; i < limit && numPending > 0; i++) {
        final ThreadState next = perThreadPool.getThreadState(i);
        if (next.flushPending) {
          final DocumentsWriterPerThread dwpt = tryCheckoutForFlush(next);
          if (dwpt != null) {
            return dwpt;
          }
        }
      }
    }
    return null;
  }

synchronized DocumentsWriterPerThread tryCheckoutForFlush(ThreadState perThread) {
   // 如果该ThreadState处于flushPending状态，尝试取出该ThreadState关联的DWPT
   return perThread.flushPending ? internalTryCheckOutForFlush(perThread) : null;
}

  private DocumentsWriterPerThread internalTryCheckOutForFlush(ThreadState perThread) {
    assert Thread.holdsLock(this);
    assert perThread.flushPending;
    try {
      // We are pending so all memory is already moved to flushBytes
      // 由于添加、更新文档是并发操作，可能其他的线程获得了该ThreadState的锁，
      // 如果无法获取该ThreadState的锁，返回空
      if (perThread.tryLock()) {
        try {
          // 如果该ThreadState中的DWPT已经初始化
          if (perThread.isInitialized()) {
            assert perThread.isHeldByCurrentThread();
            final DocumentsWriterPerThread dwpt;
            final long bytes = perThread.bytesUsed; // do that before
                                                         // replace!
            // 重置该ThreadState
            dwpt = perThreadPool.reset(perThread);
            assert !flushingWriters.containsKey(dwpt) : "DWPT is already flushing";
            // Record the flushing DWPT to reduce flushBytes in doAfterFlush
            // 记录该ThreadState关联的DWPT占用的内存，以便在doAfterFlush进行处理
            flushingWriters.put(dwpt, Long.valueOf(bytes));
            // 待flush的DWPT的个数递减
            numPending--; // write access synced
            return dwpt;
          }
        } finally {
          perThread.unlock();
        }
      }
      return null;
    } finally {
      updateStallState();
    }
  }