在访问事件文件之前等待 WatchService 事件完成处理

Question

我正在使用java.nio.file.WatchService 观看目录； 每次创建文件时，我都会调用 processFile() 并传递文件句柄。

这是观察者的代码：

boolean poll = true;
System.out.println("Watching Directory: "+path.toString()+"...");

while (poll) {
    WatchKey key = watchService.take();

    for (WatchEvent<?> event : key.pollEvents()) {
        WatchEvent<Path> ev = cast(event);
        Path dir = (Path)key.watchable();
        Path fullPath = dir.resolve(ev.context());
        File file = new File(fullPath.toString());

        System.out.println("Event kind:" + event.kind()
                + ". File affected: " + event.context() 
                + ". Full path" + fullPath);

        if(event.kind() == ENTRY_CREATE) fileProcessor.processFile(file);
    }
    poll = key.reset();
}   

这里是 processFile() 代码：

public void processFile(File file) {
    String threadName = "Thread-" + file.getName();
    Thread t = ProcessorUtil.getThreadByName(threadName);

    if (t == null || !t.isAlive()) {
        new Thread(() -> {
            try {
                Thread.sleep(1000); //HACKY WAY of WAITING
                InputStream in = new FileInputStream(file);
                someProcess(in);
            } catch (Exception e) {
                _log.error("Exception: ",e);
            }
        }, threadName).start();
    } else {
        System.out.println("File "+file.getName()+"  currently being processes");
    }

如您所见，我正在等待系统写入文件，然后才能访问FileInputStream。这对我来说似乎不对，我想知道是否有比调用Thread.sleep() 更好的方法。

Answer 1

watchservice 可以将许多事件一起发送给您，甚至可能为同一文件提供 CREATE 然后 MODIFY 事件，或在其他地方写入的文件的相同事件列表中提供 DELETE、CREATE、MODIFY 所有事件（例如，文本编辑器重写）。

防止中间进程写入和报告等到监视服务暂停的最简单方法。您的逻辑需要 ws.take() 进行第一次查询，然后 ws.poll(1) 如果您已经收到事件：

WatchKey wk = hasPending ? ws.poll(1,TimeUnit.MILLISECONDS) : ws.take();

这是来自一个测试程序的 sn-p，它整理所有事件并在最后一个 poll(1) 没有返回观察结果时对它们进行操作。

Set<Path> created = new LinkedHashSet<>();
Set<Path> modified = new LinkedHashSet<>();
Set<Path> deleted = new LinkedHashSet<>();

while(appIsRunning) {
    boolean hasPending = created.size() + modified.size() + deleted.size() > 0;
    WatchKey wk = hasPending ? ws.poll(100,TimeUnit.MILLISECONDS) : ws.take();

    if (wk != null)  {
        for (WatchEvent<?> event : wk.pollEvents()) {
             Path parent = (Path) wk.watchable();
             Path eventPath = (Path) event.context();
             store(event.kind(), parent.resolve(eventPath), created, modified, deleted);
         }
         boolean valid = wk.reset();
         if (!valid) {
             throw new RuntimeException("Check the path, it may be deleted: "+dir);
         }
    }

    System.out.println("PENDING: cre="+created.size()+" mod="+modified.size()+" del="+deleted.size());

    // ONLY HANDLE EVENTS IF NOTHING WAS RECEIVED:
    if (wk == null && hasPending) {
        // FIRE EVENTS for each list HERE: deleted, created, modified

        // reset the list for next take() cycle:
        deleted.clear();  created.clear(); modified.clear();
    }
}

如果您只捕捉 CREATE 事件，store() 很容易，此版本将多种类型整理为最新的合理类型：

/**
 * Save event for later processing by event kind EXCEPT for:
 * <li>DELETE followed by CREATE           => store as MODIFY
 * <li>CREATE followed by MODIFY           => store as CREATE
 * <li>CREATE or MODIFY followed by DELETE => store as DELETE
 */
private static void
store(Kind<?> kind, Path path, Set<Path> created, Set<Path> modified, Set<Path> deleted) {
    System.out.println("STORE "+kind+" path:"+path);

    boolean cre = false;
    boolean mod = false;
    boolean del = kind == StandardWatchEventKinds.ENTRY_DELETE;

    if (kind == StandardWatchEventKinds.ENTRY_CREATE) {
        mod = deleted.contains(path);
        cre = !mod;
    }
    else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
        cre = created.contains(path);
        mod = !cre;
    }
    addOrRemove(created,  cre, path);
    addOrRemove(modified, mod, path);
    addOrRemove(deleted,  del, path);
}

// Add or remove from the set:
private static void addOrRemove(Set<Path> set, boolean add, Path path) {
    if (add) set.add(path);
    else     set.remove(path);
}

这减少了 CREATE 的事件处理程序与写入操作发生冲突或错过文件结尾的机会。