java.nio.file.AccessMode#java.nio.file.WatchKey源码实例Demo

/**
 * Stress the given WatchService, specifically the cancel method, in
 * the given directory. Closes the WatchService when done.
 */
static void handle(Path dir, WatchService watcher) {
    try {
        try {
            Path file = dir.resolve("anyfile");
            for (int i=0; i<2000; i++) {
                WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE);
                Files.createFile(file);
                Files.delete(file);
                key.cancel();
            }
        } finally {
            watcher.close();
        }
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}
 

private void removeFileFromQueue(WatchKey key, WatchEvent<Path> pathEvent) {
  getMonitor()
      .debug(
          "ENTRY_DELETE event received - file '{}' will be removed from queue",
          pathEvent.context());
  try {
    Path dir = watchKeys.get(key);
    if (dir != null) {
      Path resolved = dir.resolve(pathEvent.context());
      queue.remove(resolved);
    } else {
      getMonitor()
          .warn(
              "WatchKey not found - file '{}' will not be removed from the queue",
              pathEvent.context());
    }
  } catch (Exception ioe) {
    getMonitor()
        .warn(
            "An error occurred - file '{}' will not be removed from the queue",
            pathEvent.context(),
            ioe);
  }

  queue.remove(pathEvent.context());
}
 

/**
 * Stress the given WatchService, specifically the cancel method, in
 * the given directory. Closes the WatchService when done.
 */
static void handle(Path dir, WatchService watcher) {
    try {
        try {
            Path file = dir.resolve("anyfile");
            for (int i=0; i<2000; i++) {
                WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE);
                Files.createFile(file);
                Files.delete(file);
                key.cancel();
            }
        } finally {
            watcher.close();
        }
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}
 

public void watch() throws IOException, InterruptedException {
  LOG.info("Watching directory {} for event(s) {}", watchDirectory, watchEvents);

  WatchKey watchKey = watchDirectory.register(
    watchService,
    watchEvents.toArray(new WatchEvent.Kind[watchEvents.size()])
  );

  while (!stopped) {
    if (watchKey != null) {
      processWatchKey(watchKey);

      if (!watchKey.reset()) {
        LOG.warn("WatchKey for {} no longer valid", watchDirectory);
        break;
      }
    }

    watchKey = watchService.poll(pollWaitCheckShutdownMillis, TimeUnit.MILLISECONDS);
  }
}
 

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}
 

/**
 * Register the given directory with the WatchService.
 *
 * @param dir the directory to register watch on
 * @throws IOException Signals that an I/O exception has occurred.
 */
private void register(Path dir) throws IOException {

    for(Path p: keys.values()) {
        // This may NOT be correct for all cases (ensure resolve will work!)
        if(dir.startsWith(p)) {
            LOGGER.debug("Path {} watched via {}", dir, p);
            return;
        }
    }

    if (FILE_TREE == null) {
        LOGGER.debug("WATCHING:ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY - high} {}", dir);
    } else {
        LOGGER.debug("WATCHING: ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY - fileTree,high {}", dir);
    }

    final WatchKey key = dir.register(watcher, KINDS,  MODIFIERS);

    keys.put(key, dir);
}
 

/**
 * Wait until a PubSub emulator configuration file is updated.
 * Fail if the file does not update after 1 second.
 * @param watchService the watch-service to poll
 * @throws InterruptedException which should interrupt the peaceful slumber and bubble up
 * to fail the test.
 */
private void updateConfig(WatchService watchService) throws InterruptedException {
	int attempts = 10;
	while (--attempts >= 0) {
		WatchKey key = watchService.poll(100, TimeUnit.MILLISECONDS);

		if (key != null) {
			Optional<Path> configFilePath = key.pollEvents().stream()
					.map((event) -> (Path) event.context())
					.filter((path) -> ENV_FILE_NAME.equals(path.toString()))
					.findAny();
			if (configFilePath.isPresent()) {
				return;
			}
		}
	}

	fail("Configuration file update could not be detected");
}
 

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}
 

public HomeDirectoryTreeWalker(final WatchService watchService, final Map<WatchKey, Path> watchKeys,
                final Path homeDirectoryBasePath, final List<Path> excludedPaths,
                final MultiUserAuthorizedKeysMap authorizedKeysMap) {

    this.excludedPaths = new ArrayList<>(excludedPaths.size());

    // make sure all paths are absolute
    for (Path excluded : excludedPaths) {
        this.excludedPaths.add(excluded.toAbsolutePath());
    }

    this.watchService = watchService;
    this.watchKeys = watchKeys;
    this.authorizedKeysMap = authorizedKeysMap;
    this.homeDirectoryBasePath = homeDirectoryBasePath.toAbsolutePath();
    this.homeDirNameCount = this.homeDirectoryBasePath.getNameCount();
}
 

@Override
protected String nextEvent() throws IOException, InterruptedException {
    WatchKey key;
    try {
        key = watcher.take();
    } catch (ClosedWatchServiceException cwse) { // #238261
        @SuppressWarnings({"ThrowableInstanceNotThrown"})
        InterruptedException ie = new InterruptedException();
        throw (InterruptedException) ie.initCause(cwse);
    }
    Path dir = (Path)key.watchable();
           
    String res = dir.toAbsolutePath().toString();
    for (WatchEvent<?> event: key.pollEvents()) {
        if (event.kind() == OVERFLOW) {
            // full rescan
            res = null;
        }
    }
    key.reset();
    return res;
}
 

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}
 

/**
 * Register all keys.
 *
 * @param directory directory to register
 */
private void registerAll(Path directory) {
  File file = directory.toFile();
  // only for unhidden directories
  if (file.isDirectory() && !file.isHidden()) {
    try {
      // create a new watch key for the directory
      WatchKey key = directory.register(
        watcher,
        StandardWatchEventKinds.ENTRY_CREATE,
        StandardWatchEventKinds.ENTRY_DELETE,
        StandardWatchEventKinds.ENTRY_MODIFY
      );
      keys.put(key, directory);
      File[] files = file.listFiles();
      if (files != null) {
        // walk directory tree
        for (File f : files) {
          registerAll(f.toPath());
        }
      }
    } catch (IOException e) {
      // skip directory and subtree :]
    }
  }
}
 

private WatchKey take() throws InterruptedException {
    synchronized(this) {
        if(interrupted) {
            interrupted = false;
            throw new InterruptedException();
        } else {
            mayInterrupt = true;
        }
    }

    try {
        return watcher.take();
    } finally {
        synchronized(this) {
            mayInterrupt = false;
        }
    }
}
 

public void runInternal() throws Exception {
    WatchKey watchKey = watcher.take();
    watchKey.reset();
    if (!certificateModified(watchKey)) {
        return;
    }
    // wait for system to settle (in case copying over cert/key pair one by one)
    while (true) {
        SECONDS.sleep(5);
        watchKey = watcher.poll();
        if (watchKey == null) {
            break;
        }
        watchKey.reset();
        if (!certificateModified(watchKey)) {
            break;
        }
    }
    delegatingSslContext.reloadCertificate();
}
 

public void run() {
  while (true) {
    try {
      WatchKey watchKey = service.take();
      // 清理掉已发生的事件，否则会导致事件遗留，进入死循环
      watchKey.pollEvents();
      synchronized (this) {
        createAuthHeaders();
      }
      watchKey.reset();
    } catch (InterruptedException e) {
      LOGGER.error("error occured. detail : {}", e.getMessage());
    }
  }
}
 

protected void watcherPid(final File jettyPidFile) throws IOException {
    final WatchService watcher = FileSystems.getDefault().newWatchService();
    final Path path = Paths.get(".");
    path.register(watcher, StandardWatchEventKinds.ENTRY_DELETE);
    
    Executors.newFixedThreadPool(1, (runnable) -> {
        final Thread jetty = new Thread(runnable);
        jetty.setName("Jetty PID Watcher: " + System.currentTimeMillis());
        return jetty;
    }).execute(() -> {
        try {
            for (;;) {
                final WatchKey watchKey = watcher.take();
                final List<WatchEvent<?>> events = watchKey.pollEvents();
                for(WatchEvent<?> event : events) {
                    final String fileName = ((Path) event.context()).toFile().getAbsolutePath();
                    if (jettyPidFile.getAbsolutePath().equals(fileName)) {
                        LOGGER.info("jetty.pid已被删除，应用进入退出流程");
                        System.exit(0);
                    }
                }
                
                watchKey.reset();
            }
        } catch (final InterruptedException e) {
            LOGGER.info("Stoped File Watcher");
        }
    });
}
 

private static void watch(final WatchKey key) {

        if (watcherThread != null) {
            // tell the old watchter thread to shutdown
            watcherThread.interrupt();
        }

         watcherThread = new Thread("ArchivingTest.watch") {
            @Override
            public void run() {


                for (; ; ) {
                    for (final WatchEvent<?> event : key.pollEvents()) {
                        final WatchEvent.Kind<?> kind = event.kind();

                        if (kind == StandardWatchEventKinds.OVERFLOW) {
                            continue;
                        }

                        if (watcherThread != this || isInterrupted()) {
                            return;
                        }

                        lastEvent.set(event);

                        latch.get().countDown();
                    }

                    final boolean valid = key.reset();
                    if (!valid) {
                        System.out.println("ArchivingTest.watch terminated");
                        break;
                    }
                }
            }
        };

        watcherThread.start();
    }
 

@Override
void implCancelKey(WatchKey obj) {
    WindowsWatchKey key = (WindowsWatchKey)obj;
    if (key.isValid()) {
        fk2key.remove(key.fileKey());
        ck2key.remove(key.completionKey());
        key.invalidate();
    }
}
 

/**
 * Every time doHasNext() is called, check the WatchService for new events and add all new events
 * to the queue. Then return true if there are files on the queue, or false otherwise.
 *
 * <p>If the event indicates that a file has been deleted, ensure it is removed from the queue.
 */
@Override
public boolean doHasNext() throws IOException, CollectionException {
  WatchKey key;
  while ((key = watcher.poll()) != null) {
    for (WatchEvent<?> event : key.pollEvents()) {
      processEvent(key, event);
      getMonitor().meter("events").mark();
    }

    key.reset();
  }

  return !currLines.isEmpty() || !queue.isEmpty();
}
 

public File takeHeapDump() throws IOException {
    if (!isTakeHeapDumpSupported()) {
        throw new UnsupportedOperationException();
    }
    String cwd = monitoredVm.findByName(USER_DIR_COUNTER_NAME);
    Path applicationCwd = Paths.get(cwd);
    WatchService watchService = FileSystems.getDefault().newWatchService();
    WatchKey key = applicationCwd.register(watchService, StandardWatchEventKinds.ENTRY_CREATE);
    Runtime.getRuntime().exec(new String[] {"kill", "-USR1", String.valueOf(application.getPid())});
    try {
        Path name = findHeapDumpFile(key);
        if (name == null) {
            key = watchService.poll(20, TimeUnit.SECONDS);
            name = findHeapDumpFile(key);
        }
        watchService.close();
        if (name == null) {
            return null;
        }
        File dumpFile = applicationCwd.resolve(name).toFile();
        waitDumpDone(dumpFile);
        return dumpFile;
    } catch (InterruptedException ex) {
        watchService.close();
        return null;
    }
}
 

@Override
void implCancelKey(WatchKey obj) {
    WindowsWatchKey key = (WindowsWatchKey)obj;
    if (key.isValid()) {
        fk2key.remove(key.fileKey());
        ck2key.remove(key.completionKey());
        key.invalidate();
    }
}
 

public void watch() {
   final WatchService watcher; 
   if (watchDir == null) {
      return;
   }
   try {
      watcher = FileSystems.getDefault().newWatchService();
      watchDir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
      logger.trace("Watching driver directory [{}]", watchDir);
   } catch (IOException ex) {
      logger.error("Could not register watcher for driver directory [{}]", watchDir);
      return;
   }
   
   executor.submit(new Runnable() {
      @Override
      public void run() {
         while (true) {
            WatchKey key = null;
            try {
               key = watcher.take();
            } catch (InterruptedException e) {
               logger.error("Interrupted Exception encountered [{}]", e);
            }
            if (key != null) {
               if (key.pollEvents().size() > 0) {
                  // Something has changed. That's all we need to know.
                  for (DriverWatcherListener listener : listeners) {
                     listener.onChange();
                  }
               }
               boolean isValid = key.reset();
               if (!isValid) {
                  logger.error("Unable to watch driver directory. Watcher key invalid.");
               }
            }
         }
      }
   });
}
 

@Override
WatchKey register(Path path,
                  WatchEvent.Kind<?>[] events,
                  WatchEvent.Modifier... modifiers)
     throws IOException
{
    // delegate to poller
    return poller.register(path, events, modifiers);
}
 

/**
 * Creates a WatchService and registers the given directory
 */
private void setupWatch(String initialText) throws IOException {
    this.watcher = FileSystems.getDefault().newWatchService();
    this.dir = Files.createTempDirectory("REPL");
    this.tmpfile = Files.createTempFile(dir, null, ".repl");
    Files.write(tmpfile, initialText.getBytes(Charset.forName("UTF-8")));
    dir.register(watcher,
            ENTRY_CREATE,
            ENTRY_DELETE,
            ENTRY_MODIFY);
    watchedThread = new Thread(() -> {
        for (;;) {
            WatchKey key;
            try {
                key = watcher.take();
            } catch (ClosedWatchServiceException ex) {
                break;
            } catch (InterruptedException ex) {
                continue; // tolerate an intrupt
            }

            if (!key.pollEvents().isEmpty()) {
                if (!input.terminalEditorRunning()) {
                    saveFile();
                }
            }

            boolean valid = key.reset();
            if (!valid) {
                errorHandler.accept("Invalid key");
                break;
            }
        }
    });
    watchedThread.start();
}
 

private void addWatcher(Path watchPath, Entry watchEntry) throws IOException
{
    if (Files.notExists(watchPath))
    {
        new NonExistingParentEntry(this, watchPath, watchEntry).handleNextEntry();
    }
    else
    {
        synchronized (mapLock)
        {
            // If the directory referenced by the FileEntry is not currently
            // being watched, register the directory and create a new entry
            // for it in the watchMap. Otherwise, add the FileEntry to the
            // list of watchers associated with the corresponding watchMap
            // entry.
            WatchMapEntry wMapEntry = watchMap.get(watchPath);
            if (wMapEntry == null)
            {
                WatchKey key = watchPath.register(
                    wSvc,
                    StandardWatchEventKinds.ENTRY_CREATE,
                    StandardWatchEventKinds.ENTRY_DELETE,
                    StandardWatchEventKinds.ENTRY_MODIFY
                );
                wMapEntry = new WatchMapEntry(key);
                watchMap.put(watchPath, wMapEntry);
            }
            wMapEntry.entries.add(watchEntry);
        }
    }
}
 

/** Returns the given key, throwing an exception if it's the poison. */
@NullableDecl
private WatchKey check(@NullableDecl WatchKey key) {
  if (key == poison) {
    // ensure other blocking threads get the poison
    queue.offer(poison);
    throw new ClosedWatchServiceException();
  }
  return key;
}
 

private static void handleEvent(final WatchKeyHolder watchKeys, final WatchKey key)
    throws IOException {
  for (final WatchEvent<?> event : key.pollEvents()) {
    if (event.kind() == StandardWatchEventKinds.OVERFLOW) {
      continue;
    }

    final WatchEvent<Path> watchEvent = cast(event);
    Path path = watchKeys.get(key);
    if (path == null) {
      continue;
    }

    path = path.resolve(watchEvent.context());
    if (Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) {
      if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE) {
        watchKeys.register(path);
      }
    } else {
      // Dispatch
      FileEvent fe = toEvent(watchEvent, path);
      if (fe != null) {
        Executor.getInstance().getEventBus().post(fe);
      }
    }
  }
}
 

private void watchThreadFunc() {
    logger.info("Config watch thread is started");
    while(state!=ServiceState.Stopped) {
        WatchKey watchKey = null;
        try{
            watchKey = watcher.poll(100, TimeUnit.MILLISECONDS);
        }catch(Throwable t) {}
        if ( watchKey==null ) {
            Thread.yield();
            continue;
        }
        ConfigProviderEntry providerEntry = getEntryByFile(watchKey);
        if ( providerEntry!=null ) {
            for(WatchEvent<?> event:watchKey.pollEvents()) {
                WatchEvent<Path> ev = (WatchEvent<Path>) event;
                WatchEvent.Kind<?> kind = event.kind();
                Path filename = ev.context();
                if (kind == java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY
                && filename.toString().equals(providerEntry.file.getName())) {
                    doReload(providerEntry);
                }
            }
        }
        watchKey.reset();
    }
    logger.info("Config watch thread is stopped");
}
 

@Override
public void run() {
    WatchService watchService = WatchServiceUtil.watchModify(confDir);
    WatchKey key;
    while (watchService != null){
        try {
            key = watchService.take();
            for (WatchEvent<?> watchEvent : key.pollEvents()) {
                if(watchEvent.kind() == ENTRY_MODIFY){
                    String fileName = watchEvent.context() == null ? "" : watchEvent.context().toString();
                    if("authorization.properties".equals(fileName)){
                        log.info("检测到授权文件authorization.properties有改动,正在重新配置相关插件配置");
                        Set<Plugin> plugins = PluginLibraryHelper.getPluginsAboutAuthorization();
                        for (Plugin plugin : plugins) {
                            plugin.init(PluginLibraryHelper.getPluginConfig(plugin));
                            log.info("已完成插件{}的配置重新加载",plugin.pluginName());
                        }
                    }
                }
            }
            key.reset();
        } catch (Exception e) {
            log.error("插件配置文件监听异常",e);
            break;
        }
    }
}
 

@SuppressWarnings("all")
private void register(Path path) throws IOException {
    WatchKey watchKey = path.register(
            watchService,
            new WatchEvent.Kind[]{
                    StandardWatchEventKinds.ENTRY_CREATE,
                    StandardWatchEventKinds.ENTRY_MODIFY,
                    StandardWatchEventKinds.ENTRY_DELETE
            });

    watchKeys.put(watchKey, path);
}