Répertoire à la recherche de modifications en Java

Cela fonctionne en mettant en œuvre un algorithme: le thread Watcher place les noms de fichiers dans BlockingQueue et un autre thread interroge cette queue, prend les noms de fichier, essaie plusieurs fois d’ouvrir le fichier. Si le fichier est ouvert, Windows Copier a libéré le locking de fichier et nous pouvons continuer. Ainsi, lorsque d’autres threads trouvent que le fichier a été déverrouillé, l’autre thread place ce nom de fichier dans la queue traitée, à partir de laquelle mon application récupère le nom de fichier. De même, l’autre thread vérifiant le délocking du fichier en ouvrant le fichier; s’il est long à débloquer un fichier, nous pouvons replacer ce nom de fichier dans BlockingQueue et traiter d’autres noms de fichiers.

Solution: J’espère que cela pourra aider d’autres:

 package dirwatch; import java.nio.file.*; import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE; import static java.nio.file.StandardWatchEventKinds.OVERFLOW; import static java.nio.file.LinkOption.*; import java.nio.file.atsortingbute.*; import java.io.*; import java.util.*; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; public class WatchDir { private final WatchService watcher; private final Map keys; private final boolean recursive; private boolean trace = false; private BlockingQueue fileProcessingQueue; //******* processedFileQueue **** will be used by other threads to resortingve unlocked files.. so I have // kept as public final public final BlockingQueue processedFileQueue; private volatile boolean closeProcessingThread; private volatile boolean closeWatcherThread; private void processFiles(){ System.out.println("DirWatchProcessingThread Started"); Ssortingng fileName; outerLoop: while(!closeProcessingThread || !fileProcessingQueue.isEmpty()){ try{ fileName = fileProcessingQueue.poll(1000, TimeUnit.MILLISECONDS); }catch(InterruptedException ie){ fileName = null; } if(fileName == null || fileName.equals("")){ continue outerLoop; } long startTime = System.currentTimeMillis(); innerLoop: while(true){ FileInputStream fis = null; File file = new File(fileName); try{ fis = new FileInputStream(fileName); break innerLoop; }catch(FileNotFoundException fnfe){ if(!file.exists() || file.isDirectory()){ System.out.println("File: '"+fileName+"has been deleted in file system or it is not file. Not processing this file."); continue outerLoop; } try{ Thread.sleep(WatchDirParameters.millisToPuaseForFileLock); }catch(InterruptedException ie){ } if((System.currentTimeMillis() - startTime) > WatchDirParameters.millisToSwapFileForUnlocking){ if(fileProcessingQueue.offer(fileName)){ continue outerLoop; }else{ startTime = System.currentTimeMillis(); continue innerLoop; } } }finally{ if(fis != null){ try{ fis.close(); }catch(IOException ioe){ ioe.printStackTrace(); } } } } System.out.println("Queuing File: "+fileName); processedLoop:while(true){ try{ if(processedFileQueue.offer(fileName, 1000, TimeUnit.MILLISECONDS)){ break processedLoop; } }catch(InterruptedException ie){ //ie.printStackTrace(); } } } closeWatcherThread = true; closeProcessingThread = true; System.out.println("DirWatchProcessingThread Exited"); } /** * Process all events for keys queued to the watcher */ private void processEvents(){ System.out.println("DirWatcherThread started."); while(!closeWatcherThread) { // wait for key to be signalled WatchKey key; try { key = watcher.take(); } catch (InterruptedException x) { // if we are returning from these method, it means we no longer wants to watch directory // we must close thread which may be waiting for file names in queue continue; }catch(ClosedWatchServiceException cwse){ break; } Path dir = keys.get(key); if (dir == null) { System.err.println("WatchKey not recognized!!"); continue; } try{ for (WatchEvent event: key.pollEvents()) { WatchEvent.Kind kind = event.kind(); if (kind == OVERFLOW) { continue; } // Context for directory entry event is the file name of entry WatchEvent ev = cast(event); Path name = ev.context(); Path child = dir.resolve(name); if(kind.equals(ENTRY_CREATE)){ // if directory is created, and watching recursively, then // register it and its sub-directories if (recursive) { try { if (Files.isDirectory(child, NOFOLLOW_LINKS)) { registerAll(child); continue; } } catch (IOException x) { // ignore to keep sample readbale } } while(true){ if(fileProcessingQueue.remainingCapacity() < 2){ // if only one last can be inserted then don't queue this we need 1 empty space in queue // for swaping file names.. // sleep for some time so processing thread may have made some rooms to queue in fileQueue // this logic will not create any problems as only one this thread is inserting in queue try{ Thread.sleep(200); }catch(InterruptedException ie){ } continue; } if(!fileProcessingQueue.offer(child.toString())){ // couldn't queue this element by whatever reason.. we will try to enqueue again by continuing loop continue; }else{ // file name has been queued in queue break; } } } } // 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()) { break; } } }catch(ClosedWatchServiceException cwse){ break; } } closeProcessingThread = true; closeWatcherThread = true; System.out.println("DirWatcherThread exited."); } public void stopWatching(){ try{ watcher.close(); }catch(IOException ioe){ } closeProcessingThread = true; closeWatcherThread = true; } public static WatchDir watchDirectory(String dirName, boolean recursive) throws InvalidPathException, IOException, Exception{ try{ Path dir = Paths.get(dirName); final WatchDir watchDir = new WatchDir(dir, recursive); watchDir.closeProcessingThread = false; watchDir.closeWatcherThread = false; new Thread(new Runnable() { public void run() { watchDir.processFiles(); } }, "DirWatchProcessingThread").start(); new Thread(new Runnable() { public void run() { watchDir.processEvents(); } }, "DirWatcherThread").start(); return watchDir; }catch(InvalidPathException ipe){ throw ipe; }catch(IOException ioe){ throw ioe; }catch(Exception e){ throw e; } } @SuppressWarnings("unchecked") private static  WatchEvent cast(WatchEvent event) { return (WatchEvent)event; } /** * Register the given directory with the WatchService */ private void register(Path dir) throws IOException { //WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY); WatchKey key = dir.register(watcher, ENTRY_CREATE); 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 void registerAll(final Path start) throws IOException { // register directory and sub-directories Files.walkFileTree(start, new SimpleFileVisitor() { @Override public FileVisitResult preVisitDirectory(Path dir, BasicFileAtsortingbutes attrs) throws IOException { register(dir); return FileVisitResult.CONTINUE; } }); } /** * Creates a WatchService and registers the given directory */ private WatchDir(Path dir, boolean recursive) throws IOException { fileProcessingQueue = new ArrayBlockingQueue(WatchDirParameters.fileQueueSize, false); processedFileQueue = new ArrayBlockingQueue(WatchDirParameters.fileQueueSize, false); this.watcher = FileSystems.getDefault().newWatchService(); this.keys = new HashMap(); this.recursive = recursive; //CreateTxtFile.createFile(dir, 1); if (recursive) { System.out.format("Scanning %s ...\n", dir); registerAll(dir); System.out.println("Done."); } else { register(dir); } // enable trace after initial registration this.trace = true; } } 

Classe de paramètre:

 package dirwatch; public class WatchDirParameters { public static final int millisToPuaseForFileLock = 200; public static final int fileQueueSize = 500; public static final int millisToSwapFileForUnlocking = 2000; } 

Création d’une version mise à jour du fichier fourni par @UDPLover, conçue pour être utilisée dans un environnement à fort taux d’access aux fichiers à l’intérieur du vérificateur de blocage de fichiers lui-même. A également créé une méthode print () qui permet d’activer ou de désactiver tout ce qui est imprimé sur la console par WatchCore. Le fichier qui interroge la boucle de l’exemple d’origine a été mis à jour pour utiliser la fonction JDK8 pour la boucle, toutes les parties étant threadées / interrompues. Ceci n’a pas encore été testé, sera mis à jour avec des corrections lorsque je pourrai le tester.

 package filewatcher; import java.io.File; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.IOException; import java.io.PrintStream; import java.nio.file.ClosedWatchServiceException; 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.Paths; import java.nio.file.SimpleFileVisitor; import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE; 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.atsortingbute.BasicFileAtsortingbutes; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; import java.util.logging.Level; import java.util.logging.Logger; /** * * @author Nackloose * http://stackoverflow.com/questions/13998379/directory-watching-for-changes-in-java */ public abstract class WatchCore extends Thread { //make class a thread by default /** * After the WatchCore recieves an event for a file and deems it unlocked, * it will be passed to this function * * @param e WatchEvent for the file, after it has been affirmed to be * unlocked. */ public abstract void onEventAndUnlocked(WatchEvent e); private final WatchService watcher; private final Map keys; private final boolean recursive; private boolean trace = false; //converted to HashMap to remove the limitation as I need this in a high rate of file access enviroment. //as well as to carry the event passed for that folder into the block check itself. //got rid of the finished queue and made events pass to the abstract void above private final HashMap fileProcessingQueue; //create a varible to keep track of the thread checking the file blocking, so we can start and stop it. private final WatchBlocker blocker; public WatchCore(Ssortingng dir) throws IOException { //defaultly dont recurse this(dir, false); } public WatchCore(Ssortingng dir, boolean recursive) throws IOException { this(Paths.get(dir), recursive); } public WatchCore(Path dir) throws IOException { //defaultly dont recurse this(dir, false); } public WatchCore(Path dir, boolean recursive) throws IOException { fileProcessingQueue = new HashMap<>(); this.watcher = FileSystems.getDefault().newWatchService(); this.keys = new HashMap<>(); this.recursive = recursive; //CreateTxtFile.createFile(dir, 1); if (recursive) { print("Scanning %s ...", dir); registerAll(dir); print("Done."); } else { register(dir); } // enable trace after initial registration this.trace = true; //start the thread to process files to be checked for file blocking blocker = new WatchBlocker(); } @SuppressWarnings("unchecked") private static  WatchEvent cast(WatchEvent event) { return (WatchEvent) event; } @Override public synchronized void start() { //start up our thread _FIRST_ super.start(); //then start the blocking thread blocker.start(); } @Override public void interrupt() { //Everything backwards, stop the blocker _FIRST_ blocker.interrupt(); //then stop our thread. super.interrupt(); } /** * Register the given directory with the WatchService */ private void register(Path dir) throws IOException { //WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY); WatchKey key = dir.register(watcher, ENTRY_CREATE); if (trace) { Path prev = keys.get(key); if (prev == null) { print("register: %s\n", dir); } else { if (!dir.equals(prev)) { print("update: %s -> %s\n", prev, dir); } } } keys.put(key, dir); } /** * Register the given directory, and all its sub-directories, with the * WatchService. */ private void registerAll(final Path start) throws IOException { // register directory and sub-directories Files.walkFileTree(start, new SimpleFileVisitor() { @Override public FileVisitResult preVisitDirectory(Path dir, BasicFileAtsortingbutes attrs) throws IOException { register(dir); return FileVisitResult.CONTINUE; } }); } /** * Process all events for keys queued to the watcher */ @Override public void run() { //this was previous called processEvents() //pruned any un-nessicary continues, labels, and labels on breaks, a lot of them //were redundant print("DirWatcherThread started."); //as long as we're not interrupted we keep working while (!interrupted()) { // wait for key to be signalled WatchKey key; try { key = watcher.take(); } catch (InterruptedException x) { // if we are returning from these method, it means we no longer wants to watch directory // we must close thread which may be waiting for file names in queue continue; } catch (ClosedWatchServiceException cwse) { break; } Path dir = keys.get(key); if (dir == null) { printe("WatchKey not recognized!!"); continue; } try { //converted to functional for loop. key.pollEvents().stream().filter((event) -> { WatchEvent.Kind kind = event.kind(); return !(kind == OVERFLOW); //make sure we do the filter }).forEach((event) -> { WatchEvent.Kind kind = event.kind(); // Context for directory entry event is the file name of entry WatchEvent ev = cast(event); Path name = ev.context(); Path child = dir.resolve(name); if (kind.equals(ENTRY_CREATE)) { // if directory is created, and watching recursively, then // register it and its sub-directories if (recursive) { try { if (Files.isDirectory(child, NOFOLLOW_LINKS)) { registerAll(child); return; //continue; } } catch (IOException x) { // ignore to keep sample readbale } } fileProcessingQueue.put(child.toSsortingng(), ev); } }); // 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()) { break; } } } catch (ClosedWatchServiceException cwse) { break; } } print("DirWatcherThread exited."); } /** * * @author * http://stackoverflow.com/questions/13998379/directory-watching-for-changes-in-java * Nackloose */ private class WatchBlocker extends Thread { @Override public synchronized void start() { //get it going super.start(); } @Override public void interrupt() { //interupt our thread super.interrupt(); } @Override public void run() { //this was perviously processFiles() //pruned any un-nessicary continues, labels, and labels on breaks, a lot of them //were redundant print("DirWatchProcessingThread Started"); Entry fileEvent; outerLoop: //as long as we're not interrupted we keep working while (!interrupted()) { if (fileProcessingQueue.isEmpty()) { try { Thread.sleep(WatchCoreParameters.timeToIdle); } catch (InterruptedException ex) { Logger.getLogger(WatchCore.class.getName()).log(Level.SEVERE, null, ex); } continue; } fileEvent = fileProcessingQueue.entrySet().iterator().next(); fileProcessingQueue.remove(fileEvent.getKey()); long startTime = System.currentTimeMillis(); while (true) { FileInputStream fis = null; File file = new File(fileEvent.getKey()); try { fis = new FileInputStream(fileEvent.getKey()); break; } catch (FileNotFoundException fnfe) { if (!file.exists() || file.isDirectory()) { print("File: '" + fileEvent + "has been deleted in file system or it is not file. Not processing this file."); continue outerLoop; } try { Thread.sleep(WatchCoreParameters.millisToPauseForFileLock); } catch (InterruptedException ie) { } if ((System.currentTimeMillis() - startTime) > WatchCoreParameters.millisToSwapFileForUnlocking) { fileProcessingQueue.put(fileEvent.getKey(), fileEvent.getValue()); } } finally { if (fis != null) { try { fis.close(); } catch (IOException ioe) { ioe.printStackTrace(); } } } } print("Queuing File: " + fileEvent); //pass the unlocked file event to the abstract method onEventAndUnlocked(fileEvent.getValue()); } print("DirWatchProcessingThread Exited"); } } /** * * @author * http://stackoverflow.com/questions/13998379/directory-watching-for-changes-in-java * Nackloose */ public static class WatchCoreParameters { public static int timeToIdle = 2000, millisToPauseForFileLock = 200, millisToSwapFileForUnlocking = 2000; public static boolean verbose = false; } // private void print(Ssortingng s) { //defaultly we're not writing an error print(s, false); } public static final void print(Ssortingng s, boolean error) { //check verbosity, exit if none. if (!WatchCoreParameters.verbose) { return; } //if this is an error, assign System.err to a temp varible //otherise assign System.out for normal printing PrintStream out = (!error ? System.out : System.err); if (s.contains("\n")) { // check to see if theirs a new line out.print(s); //print accordingly } else { out.println(s); //print accordingly } } public static final void printe(Ssortingng s) { //shortcut/convenience method for printing an error print(s, true); } public static final void print(Ssortingng s, Object... formatObj) { //check verbosity, exit if none. if (!WatchCoreParameters.verbose) { return; } //format the object into the ssortingng, and if no newline is there, add it. System.out.format(s + (s.contains("\n") ? "" : "\n"), formatObj); } // }