问题：

Java WatchService，使用线程对事件执行操作

阴雪风

2023-03-14

我可以通过使用WatchKey注册cw来监视目录（网络上有很多例子），但是这个监视程序会捕捉到每一个事件。例如，在windows上，如果我监视d:/temp dir并创建一个新的。txt文件并将其重命名，我得到以下事件。

ENTRY_CREATE: d:\temp\test\New Text Document.txt
ENTRY_MODIFY: d:\temp\test
ENTRY_DELETE: d:\temp\test\New Text Document.txt
ENTRY_CREATE: d:\temp\test\test.txt
ENTRY_MODIFY: d:\temp\test

我想在创建或更新新文件时执行一个操作。但是，我不希望在上面的示例中该操作运行5次。

我的第一个想法：因为我只需要每隔一段时间运行一次操作（在这种情况下是推送到私有Git服务器）（例如，仅当监控目录发生更改时每10秒检查一次，然后才执行推送）我想拥有一个带有布尔参数的对象，我可以从单独的线程中获取和设置。

现在，这可以正常工作了（除非大师可以帮助我了解为什么这是一个糟糕的想法），问题是，如果在SendToGit线程的操作过程中捕获了一个文件事件，并且该操作完成，那么它会将“Found”参数设置为false。紧接着捕捉到另一个事件（如上例所示），它们将再次将“Found”参数设置为true。这并不理想，因为我将立即再次运行SendToGit操作，这将是不必要的。

我的第二个想法调查暂停检查MonitorFolder线程中的更改，直到SendToGit操作完成（即继续检查ChangesFinding参数是否已设置回false。当此参数为false时，再次开始检查更改。

问题

这是一条可以接受的路，还是我已经掉进了一个没有希望回来的兔子洞？
如果我继续我的第二个想法，如果我忙于SendToGit操作并且在监控文件夹中进行了更改，会发生什么？我怀疑这不会被识别，我可能会错过更改。

剩下的代码

变化基金会。JAVA

package com.acme;

public class ChangesFound {
    
    private boolean found = false;

    public boolean wereFound() {
        return this.found;
    }

    public void setFound(boolean commitToGit) {
        this.found = commitToGit;
    }
}

在我的主应用程序中，我启动了两个线程。

监视器文件夹。java监视目录，当发现观察者事件时，将ChangesFound变量“found”设置为true

以下是我启动线程的应用程序：

package com.acme;

import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;

public class App {

    private static ChangesFound chg;
    
    public static void main(String[] args) throws IOException {

        String dirToMonitor = "D:/Temp";
        boolean recursive = true;
        chg = new ChangesFound();

        Runnable r = new SendToGit(chg);
        new Thread(r).start();

        Path dir = Paths.get(dirToMonitor);
        Runnable m = new MonitorFolder(chg, dir, recursive);
        new Thread(m).start();        
    }
}

SendToGit.java

package com.acme;

public class SendToGit implements Runnable {

    private ChangesFound changes;

    public SendToGit(ChangesFound chg) {
        changes = chg;
    }
    
    public void run() {

        while (true) {           
            try {
                Thread.sleep(10000);
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }

            System.out.println(java.time.LocalDateTime.now() + " [SendToGit] waking up.");

            if (changes.wereFound()) {
                System.out.println("\t***** CHANGES FOUND push to Git.");
                changes.setFound(false);
            } else {
                System.out.println("\t***** Nothing changed.");
            }
        }
    }
}

MonitorFolder.java（为长时间的课程道歉，我只是在这里添加了这个，以防它对其他人有帮助。）

package com.acme;

import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import static java.nio.file.LinkOption.NOFOLLOW_LINKS;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE;
import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
import java.nio.file.WatchEvent;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.HashMap;
import java.util.Map;

public class MonitorFolder implements Runnable  {
    
    private static WatchService watcher;
    private static Map<WatchKey, Path> keys;
    private static boolean recursive;
    private static boolean trace = false;
    private static boolean commitGit = false;
    private static ChangesFound changes;

    @SuppressWarnings("unchecked")
    static <T> WatchEvent<T> cast(WatchEvent<?> event) {
        return (WatchEvent<T>) event;
    }

    /**
     * Creates a WatchService and registers the given directory
     */
    MonitorFolder(ChangesFound chg, Path dir, boolean rec) throws IOException {
        changes = chg;
        watcher = FileSystems.getDefault().newWatchService();
        keys = new HashMap<WatchKey, Path>();
        recursive = rec;

        if (recursive) {
            System.out.format("[MonitorFolder] Scanning %s ...\n", dir);
            registerAll(dir);
            System.out.println("Done.");
        } else {
            register(dir);
        }

        // enable trace after initial registration
        this.trace = true;
    }

    /**
     * Register the given directory with the WatchService
     */
    private static void register(Path dir) throws IOException {
        WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
        if (trace) {
            Path prev = keys.get(key);
            if (prev == null) {
                System.out.format("register: %s\n", dir);
            } else {
                if (!dir.equals(prev)) {
                    System.out.format("update: %s -> %s\n", prev, dir);
                }
            }
        }
        keys.put(key, dir);
    }

    /**
     * Register the given directory, and all its sub-directories, with the
     * WatchService.
     */
    private static void registerAll(final Path start) throws IOException {
        // register directory and sub-directories
        Files.walkFileTree(start, new SimpleFileVisitor<Path>() {
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs)
                    throws IOException {
                register(dir);
                return FileVisitResult.CONTINUE;
            }
        });
    }

    /**
     * Process all events for keys queued to the watcher
     */
    public void run() {
        for (;;) {

            // wait for key to be signalled
            WatchKey key;
            try {
                key = watcher.take();
            } catch (InterruptedException x) {
                return;
            }

            Path dir = keys.get(key);
            if (dir == null) {
                System.err.println("WatchKey not recognized!!");
                continue;
            }

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

                // TBD - provide example of how OVERFLOW event is handled
                if (kind == OVERFLOW) {
                    System.out.println("Something about Overflow");
                    continue;
                }

                // Context for directory entry event is the file name of entry
                WatchEvent<Path> ev = cast(event);
                Path name = ev.context();
                Path child = dir.resolve(name);

                // print out event and set ChangesFound object Found parameter to True
                System.out.format("[MonitorFolder] " + java.time.LocalDateTime.now() + " - %s: %s\n", event.kind().name(), child);
                changes.setFound(true);

                // if directory is created, and watching recursively, then
                // register it and its sub-directories
                if (recursive && (kind == ENTRY_CREATE)) {
                    try {
                        if (Files.isDirectory(child, NOFOLLOW_LINKS)) {
                            registerAll(child);
                        }
                    } catch (IOException x) {
                        // ignore to keep sample readbale
                    }
                }
            }

            // reset key and remove from set if directory no longer accessible
            boolean valid = key.reset();
            if (!valid) {
                keys.remove(key);

                // all directories are inaccessible
                if (keys.isEmpty()) {
                    System.out.println("keys.isEmpty");
                    break;
                }
            }
        }
    }
}

共有1个答案

晏志明

2023-03-14

这两种策略都会导致问题，因为Watch服务非常冗长，在下游处理实际需要一两条消息时会发送许多消息，因此有时您可能会做不必要的工作或错过事件。

使用WatchService时，您可以将多个通知整理在一起，并作为一个事件传递，列出一组最近的删除、创建和更新：

删除，然后创建=

不要调用WatchService.take（）并对每条消息进行操作，而是使用WatchService.poll（timeout），并且只有在没有返回任何内容时，才会将之前的事件集的联合作为一个-而不是在每次成功轮询后单独进行。

将问题分解为两个组件更容易，这样下次需要时就不会重复WatchService代码：

一个监视管理器，它处理监视服务目录注册并整理副本以作为一个组发送给事件监听器
一个Listener类，用于接收更改组并对集合进行操作。

此示例可能有助于说明-请参阅WatchExample，它是一个管理器，用于设置注册，但向setListener定义的回调传递的事件要少得多。您可以设置MonitorFolder，比如WatchExample，以减少发现的事件，并将您在SendToGit中的代码设置为侦听器，通过聚合的文件更改集（删除、创建、更新）按需调用。

public static void main(String[] args) throws IOException, InterruptedException {

    final List<Path> dirs = Arrays.stream(args).map(Path::of).map(Path::toAbsolutePath).collect(Collectors.toList());
    Kind<?> [] kinds = { StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_MODIFY, StandardWatchEventKinds.ENTRY_DELETE};

    // Should launch WatchExample PER Filesystem:
    WatchExample w = new WatchExample();
    w.setListener(WatchExample::fireEvents);

    for(Path dir : dirs)
        w.register(kinds, dir);

    // For 2 or more WatchExample use: new Thread(w[n]::run).start();
    w.run();
}

public class WatchExample implements Runnable {

    private final Set<Path> created = new LinkedHashSet<>();
    private final Set<Path> updated = new LinkedHashSet<>();
    private final Set<Path> deleted = new LinkedHashSet<>();

    private volatile boolean appIsRunning = true;
    // Decide how sensitive the polling is:
    private final int pollmillis = 100;
    private WatchService ws;

    private Listener listener = WatchExample::fireEvents;

    @FunctionalInterface
    interface Listener
    {
        public void fileChange(Set<Path> deleted, Set<Path> created, Set<Path> modified);
    }

    WatchExample() {
    }
    
    public void setListener(Listener listener) {
        this.listener = listener;
    }

    public void shutdown() {
        System.out.println("shutdown()");
        this.appIsRunning = false;
    }

    public void run() {
        System.out.println();
        System.out.println("run() START watch");
        System.out.println();

        try(WatchService autoclose = ws) {

            while(appIsRunning) {

                boolean hasPending = created.size() + updated.size() + deleted.size() > 0;
                System.out.println((hasPending ? "ws.poll("+pollmillis+")" : "ws.take()")+" as hasPending="+hasPending);

                // Use poll if last cycle has some events, as take() may block
                WatchKey wk = hasPending ? ws.poll(pollmillis,TimeUnit.MILLISECONDS) : ws.take();
                if (wk != null)  {
                    for (WatchEvent<?> event : wk.pollEvents()) {
                         Path parent = (Path) wk.watchable();
                         Path eventPath = (Path) event.context();
                         storeEvent(event.kind(), parent.resolve(eventPath));
                     }
                     boolean valid = wk.reset();
                     if (!valid) {
                         System.out.println("Check the path, dir may be deleted "+wk);
                     }
                }

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

                // This only sends new notifications when there was NO event this cycle:
                if (wk == null && hasPending) {
                    listener.fileChange(deleted, created, updated);
                    deleted.clear();
                    created.clear();
                    updated.clear();
                }
            }
        }
        catch (InterruptedException e) {
            System.out.println("Watch was interrupted, sending final updates");
            fireEvents(deleted, created, updated);
        }
        catch (IOException e) {
            throw new UncheckedIOException(e);
        }

        System.out.println("run() END watch");
    }

    public void register(Kind<?> [] kinds, Path dir) throws IOException {
        System.out.println("register watch for "+dir);

        // If dirs are from different filesystems WatchService will give errors later
        if (this.ws == null) {
            ws = dir.getFileSystem().newWatchService();
        }
        dir.register(ws, kinds);
    }

    /**
     * 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 void
    storeEvent(Kind<?> kind, Path path) {
        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(updated, 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);
    }

    public static void fireEvents(Set<Path> deleted, Set<Path> created, Set<Path> modified) {
        System.out.println();
        System.out.println("fireEvents START");
        for (Path path : deleted)
            System.out.println("  DELETED: "+path);
        for (Path path : created)
            System.out.println("  CREATED: "+path);
        for (Path path : modified)
            System.out.println("  UPDATED: "+path);
        System.out.println("fireEvents END");
        System.out.println();
    }
}

类似资料：

对事务提交执行操作

我们使用HiberNate作为JPA提供程序。当其中一个实体更新时，我需要对Quartz计划执行一些更新。目前，该代码是在该实体的更新方法中调用的。但是，Quartz更改只有在事务成功提交时才会生效。我考虑过实现一个实体监听器，但是我只想在实体被特定方法修改时执行这些更新，并且我不确定JPA实体监听器是否支持依赖注入，我需要依赖注入来获取对Quartz调度器的引用。有没有办法以编程方式附加活动
对列执行条件操作

我知道这应该很简单，但是我想从熊猫数据框中取一列，并且只对满足某些条件（比如小于1）的条目乘以标量（比如2）。例如，在这个数据框中，如果我有兴趣在列上执行此操作，结果应该是我有以下绝对任务：但是我不知道如何使用中的实际值。提前谢谢！
多线程和使用事件

线程3：
使用Java8[duplicate]对代码执行操作

我有一个方法，它迭代一个映射，对值执行一个操作，并填充一个要返回的映射。我的问题是，我如何将其转换为Java8（执行不循环的操作）？代码：
如何使用DocumentFile API对文件执行基本操作？

从API 21（Lollipop）开始，应用程序可以获得修改真实SD卡的特殊权限，如我写的历史发文所示（这里和这里）。我可以删除文件，也可以创建文件，但我找不到执行其他基本文件操作的方法：读，写，使用InputStream和OutputStream 移动文件。创建一个文件夹而不仅仅是一个文件重命名文件获取文件信息（最近更新等）通过其他应用程序共享/打开文件。其他行动我可能已经忘记了。
使用Spring Boot执行器监视tomcat线程和线程池执行器

我们正在对使用SpringBoot 2.2.2和Spring执行器的应用程序进行性能测试。我们希望监控：正在使用多少tomcat线程有多少tomcat请求正在排队正在使用多少个ThreadPoolTaskExector线程（我们将@Async与线程池一起用于某些任务）执行器中是否提供此信息？我看不到需要使用哪些指标。

Java WatchService，使用线程对事件执行操作

共有1个答案

相关问答

相关文章

相关阅读

相关工具

相关文档