企业搜索引擎开发之连接器connector（二十八）

/**
 * Management interface to {@link DocumentSnapshotRepositoryMonitor} threads.
 *
 * @since 2.8
 */
public interface DocumentSnapshotRepositoryMonitorManager {
  /**
   * Ensures all monitor threads are running.
   *
   * @param checkpoint for the last completed document or null if none have
   *        been completed.
   * @throws RepositoryException
   */
  void start(String checkpoint) throws RepositoryException;

  /**
   * Stops all the configured {@link DocumentSnapshotRepositoryMonitor} threads.
   */
  void stop();

  /**
   * Removes persisted state for {@link DocumentSnapshotRepositoryMonitor}
   * threads. After calling this {@link DocumentSnapshotRepositoryMonitor}
   * threads will no longer be able to resume from where they left off last
   * time.
   */
  void clean();

  /**
   * Returns the number of {@link DocumentSnapshotRepositoryMonitor} threads
   * that are alive. This method is for testing purposes.
   */
  int getThreadCount();

  /**
   * Returns the {@link CheckpointAndChangeQueue} for this
   * {@link DocumentSnapshotRepositoryMonitorManager}
   */
  CheckpointAndChangeQueue getCheckpointAndChangeQueue();

  /** Returns whether we are after a start() call and before a stop(). */
  boolean isRunning();

  /**
   * Receives information specifying what is guaranteed to be delivered to GSA.
   * Every entry in passed in Map is a monitor name and MonitorCheckpoint.
   * The monitor of that name can expect that all documents before and including
   * document related with MonitorCheckpoint will be delivered to GSA.
   * This information is for the convenience and efficiency of the Monitor so
   * that it knows how many changes it has to resend.  It‘s valid for a monitor
   * to ignore these updates if it feels like it for some good reason.
   * FileConnectorSystemMonitor instances use this information to trim their
   * file system snapshots.
   */
  void acceptGuarantees(Map<String, MonitorCheckpoint> guarantees);

  /**
   * Receives {@link TraversalSchedule} from TraversalManager which is
   * {@link TraversalScheduleAware}.  
   */
  void setTraversalSchedule(TraversalSchedule traversalSchedule);
}

private volatile TraversalSchedule traversalSchedule;

//监控器线程
  private final List<Thread> threads =
      Collections.synchronizedList(new ArrayList<Thread>());
  //监控器映射容器
  private final Map<String, DocumentSnapshotRepositoryMonitor> fileSystemMonitorsByName =
      Collections.synchronizedMap(new HashMap<String, DocumentSnapshotRepositoryMonitor>());
  private boolean isRunning = false;  // Monitor threads start in off state.
  private final List<? extends SnapshotRepository<? extends DocumentSnapshot>>
      repositories;

  private final File snapshotDir;
  private final ChecksumGenerator checksumGenerator;
  //CheckpointAndChange对象容器（List）
  private final CheckpointAndChangeQueue checkpointAndChangeQueue;
  //Change对象容器（阻塞队列）
  private final ChangeQueue changeQueue;

  private final DocumentSnapshotFactory documentSnapshotFactory;

/**
   * Constructs {@link DocumentSnapshotRepositoryMonitorManagerImpl}
   * for the {@link DiffingConnector}.
   *
   * @param repositories a {@code List} of {@link SnapshotRepository
   *        SnapshotRepositorys}
   * @param documentSnapshotFactory a {@link DocumentSnapshotFactory}
   * @param snapshotDir directory to store {@link SnapshotRepository}
   * @param checksumGenerator a {@link ChecksumGenerator} used to
   *        detect changes in a document‘s content
   * @param changeQueue a {@link ChangeQueue}
   * @param checkpointAndChangeQueue a
   *        {@link CheckpointAndChangeQueue}
   */
  public DocumentSnapshotRepositoryMonitorManagerImpl(
      List<? extends SnapshotRepository<
          ? extends DocumentSnapshot>> repositories,
      DocumentSnapshotFactory documentSnapshotFactory,
      File snapshotDir, ChecksumGenerator checksumGenerator,
      ChangeQueue changeQueue,
      CheckpointAndChangeQueue checkpointAndChangeQueue) {
    this.repositories = repositories;
    this.documentSnapshotFactory = documentSnapshotFactory;
    this.snapshotDir = snapshotDir;
    this.checksumGenerator = checksumGenerator;
    this.changeQueue = changeQueue;
    this.checkpointAndChangeQueue = checkpointAndChangeQueue;
  }

/**
   * 启动方法
   */
  /** Go from "cold" to "warm" including CheckpointAndChangeQueue. */
  public void start(String connectorManagerCheckpoint)
      throws RepositoryException {

    try {
        //启动 获取Change(主要动作：从json格式队列文件加载monitorPoints和checkpointAndChangeList队列)
      checkpointAndChangeQueue.start(connectorManagerCheckpoint);
    } catch (IOException e) {
      throw new RepositoryException("Failed starting CheckpointAndChangeQueue.",
          e);
    }
    //MonitorCheckpoint容器
    Map<String, MonitorCheckpoint> monitorPoints
        = checkpointAndChangeQueue.getMonitorRestartPoints();
    
  
    Map<String, SnapshotStore> snapshotStores = null;
    
    //加载monitorName与SnapshotStore映射容器
    try {
      snapshotStores =
          recoverSnapshotStores(connectorManagerCheckpoint, monitorPoints);     
      
    } catch (SnapshotStoreException e) {
      throw new RepositoryException("Snapshot recovery failed.", e);
    } catch (IOException e) {
      throw new RepositoryException("Snapshot recovery failed.", e);
    } catch (InterruptedException e) {
      throw new RepositoryException("Snapshot recovery interrupted.", e);
    }   
    
    //启动监控线程
    startMonitorThreads(snapshotStores, monitorPoints);    
    
    isRunning = true;
  }

 /* For each start path gets its monitor recovery files in state were monitor
   * can be started. */
  /**
   * 加载monitorName与SnapshotStore映射容器
   * @param connectorManagerCheckpoint
   * @param monitorPoints
   * @return
   * @throws IOException
   * @throws SnapshotStoreException
   * @throws InterruptedException
   */
  private Map<String, SnapshotStore> recoverSnapshotStores(
      String connectorManagerCheckpoint, Map<String,
      MonitorCheckpoint> monitorPoints)
      throws IOException, SnapshotStoreException, InterruptedException {
    Map<String, SnapshotStore> snapshotStores =
        new HashMap<String, SnapshotStore>();
    for (SnapshotRepository<? extends DocumentSnapshot> repository
        : repositories) {
      String monitorName = makeMonitorNameFromStartPath(repository.getName());
      File dir = new File(snapshotDir,  monitorName);

      boolean startEmpty = (connectorManagerCheckpoint == null)
          || (!monitorPoints.containsKey(monitorName));
      if (startEmpty) {
        LOG.info("Deleting " + repository.getName()
            + " global checkpoint=" + connectorManagerCheckpoint
            + " monitor checkpoint=" + monitorPoints.get(monitorName));
        //删除该快照目录
        delete(dir);
      } else {
          //修复该快照目录
        SnapshotStore.stitch(dir, monitorPoints.get(monitorName),
            documentSnapshotFactory);
      }

      SnapshotStore snapshotStore = new SnapshotStore(dir,
          documentSnapshotFactory);

      snapshotStores.put(monitorName, snapshotStore);
    }
    return snapshotStores;
  }

 /**
   * 启动监控线程（貌似MonitorCheckpoint与SnapshotStore与monitor有映射关系）
   * Creates a {@link DocumentSnapshotRepositoryMonitor} thread for each
   * startPath.
   *
   * @throws RepositoryDocumentException if any of the threads cannot be
   *         started.
   */
  private void startMonitorThreads(Map<String, SnapshotStore> snapshotStores,
      Map<String, MonitorCheckpoint> monitorPoints)
      throws RepositoryDocumentException {

    for (SnapshotRepository<? extends DocumentSnapshot> repository
            : repositories) {
      String monitorName = makeMonitorNameFromStartPath(repository.getName());
      //monitorName snapshotStores映射
      //快照存储器（读写器）
      SnapshotStore snapshotStore = snapshotStores.get(monitorName);
      //创建监控线程
      Thread monitorThread = newMonitorThread(repository, snapshotStore,
          monitorPoints.get(monitorName));
      threads.add(monitorThread);

      LOG.info("starting monitor for <" + repository.getName() + ">");
      monitorThread.setName(repository.getName());
      monitorThread.setDaemon(true);
      monitorThread.start();
    }
  }

/**
   * 创建监控线程
   * Creates a {@link DocumentSnapshotRepositoryMonitor} thread for the provided
   * folder.
   *
   * @throws RepositoryDocumentException if {@code startPath} is not readable,
   *         or if there is any problem reading or writing snapshots.
   */
  private Thread newMonitorThread(
      SnapshotRepository<? extends DocumentSnapshot> repository,
      SnapshotStore snapshotStore, MonitorCheckpoint startCp)
      throws RepositoryDocumentException {
      //注意monitorName
    String monitorName = makeMonitorNameFromStartPath(repository.getName());
    //document在监控线程里面处理
    DocumentSnapshotRepositoryMonitor monitor =
        new DocumentSnapshotRepositoryMonitor(monitorName, repository,
            snapshotStore, changeQueue.newCallback(), DOCUMENT_SINK, startCp,
            documentSnapshotFactory);
    monitor.setTraversalSchedule(traversalSchedule);
    LOG.fine("Adding a new monitor for " + monitorName + ": " + monitor);
    fileSystemMonitorsByName.put(monitorName, monitor);
    return new Thread(monitor);
  }

/**
   * 停止监控器
   */
  private void flagAllMonitorsToStop() {
    for (SnapshotRepository<? extends DocumentSnapshot> repository
        : repositories) {
      String monitorName = makeMonitorNameFromStartPath(repository.getName());
      DocumentSnapshotRepositoryMonitor
          monitor = fileSystemMonitorsByName.get(monitorName);
      if (null != monitor) {
        monitor.shutdown();
      }
      else {
        LOG.fine("Unable to stop non existent monitor thread for "
            + monitorName);
      }
    }
  }
  /**
   * 停止监控器线程
   */
  /* @Override */
  public synchronized void stop() {
    for (Thread thread : threads) {
      thread.interrupt();
    }
    for (Thread thread : threads) {
      try {
        thread.join(MAX_SHUTDOWN_MS);
        if (thread.isAlive()) {
          LOG.warning("failed to stop background thread: " + thread.getName());
        }
      } catch (InterruptedException e) {
        // Mark this thread as interrupted so it can be dealt with later.
        Thread.currentThread().interrupt();
      }
    }
    threads.clear();

    /* in case thread.interrupt doesn‘t stop monitors */
    flagAllMonitorsToStop();

    fileSystemMonitorsByName.clear();
    changeQueue.clear();
    this.isRunning = false;
  }

 /* The monitor should exit voluntarily if set to false */
  private volatile boolean isRunning = true;