Java: просмотр каталога для перемещения больших файлов

Question

Я писал программу, которая следит за каталогом, и когда в нем создаются файлы, он меняет имя и перемещает их в новый каталог.В моей первой реализации я использовал Java Watch API API, который работал нормально, когда я тестировал файлы размером 1 КБ.Проблема, которая возникла, заключается в том, что на самом деле создаваемые файлы имеют размер от 50 до 300 Мб.Когда это произошло, API наблюдателя сразу же нашел файл, но не смог переместить его, потому что он все еще записывался.Я попытался поместить наблюдатель в цикл (который генерировал исключения, пока файл не мог быть перемещен), но это казалось довольно неэффективным.

Поскольку это не сработало, я попытался использовать таймер, который проверяет папку каждые 10 секунда затем перемещает файлы, когда это возможно.Это метод, который я использовал в конечном итоге.

Вопрос: Есть ли какой-либо сигнал для завершения записи файла без проверки исключений или постоянного сравнения размера?Мне нравится идея использовать API-интерфейс Watcher только один раз для каждого файла вместо постоянной проверки с помощью таймера (и работы с исключениями).

Все ответы приветствуются!

nt

Answer 1

Я столкнулся с той же проблемой сегодня.В моем случае небольшая задержка перед тем, как файл был фактически импортирован, не была большой проблемой, и я все еще хотел использовать API NIO2.Я выбрал решение подождать, пока файл не будет изменен в течение 10 секунд, прежде чем выполнять какие-либо операции с ним.

Важная часть реализации заключается в следующем.Программа ожидает, пока не истечет время ожидания или не произойдет новое событие.Время истечения сбрасывается каждый раз, когда файл изменяется.Если файл удаляется до истечения времени ожидания, он удаляется из списка.Я использую метод poll с тайм-аутом ожидаемого времени истечения, то есть (lastmodified + waitTime) -currentTime

private final Map<Path, Long> expirationTimes = newHashMap();
private Long newFileWait = 10000L;

public void run() {
    for(;;) {
        //Retrieves and removes next watch key, waiting if none are present.
        WatchKey k = watchService.take();

        for(;;) {
            long currentTime = new DateTime().getMillis();

            if(k!=null)
                handleWatchEvents(k);

            handleExpiredWaitTimes(currentTime);

            // If there are no files left stop polling and block on .take()
            if(expirationTimes.isEmpty())
                break;

            long minExpiration = min(expirationTimes.values());
            long timeout = minExpiration-currentTime;
            logger.debug("timeout: "+timeout);
            k = watchService.poll(timeout, TimeUnit.MILLISECONDS);
        }
    }
}

private void handleExpiredWaitTimes(Long currentTime) {
    // Start import for files for which the expirationtime has passed
    for(Entry<Path, Long> entry : expirationTimes.entrySet()) {
        if(entry.getValue()<=currentTime) {
            logger.debug("expired "+entry);
            // do something with the file
            expirationTimes.remove(entry.getKey());
        }
    }
}

private void handleWatchEvents(WatchKey k) {
    List<WatchEvent<?>> events = k.pollEvents();
    for (WatchEvent<?> event : events) {
        handleWatchEvent(event, keys.get(k));
    }
    // reset watch key to allow the key to be reported again by the watch service
    k.reset();
}

private void handleWatchEvent(WatchEvent<?> event, Path dir) throws IOException {
    Kind<?> kind = event.kind();

    WatchEvent<Path> ev = cast(event);
        Path name = ev.context();
        Path child = dir.resolve(name);

    if (kind == ENTRY_MODIFY || kind == ENTRY_CREATE) {
        // Update modified time
        FileTime lastModified = Attributes.readBasicFileAttributes(child, NOFOLLOW_LINKS).lastModifiedTime();
        expirationTimes.put(name, lastModified.toMillis()+newFileWait);
    }

    if (kind == ENTRY_DELETE) {
        expirationTimes.remove(child);
    }
}

Answer 2

Запись другого файла в качестве указания на завершение исходного файла. I.g 'fileorg.dat' увеличивается, если готово, создайте файл 'fileorg.done' и проверьте только для файла fileorg.done.

С умными соглашениями об именах у вас не должно быть проблем.

Answer 3

Два решения:

Первое - небольшое отклонение ответа от укладчика :

Используйте уникальный префикс для неполных файлов.Что-то вроде myhugefile.zip.inc вместо myhugefile.zip.Переименуйте файлы после завершения загрузки / создания.Исключите файлы .inc из часов.

Вторым является использование другой папки на том же диске для создания / загрузки / записи файлов и перемещения их в просматриваемую папку, как только они будут готовы.Перемещение должно быть атомарным действием, если они находятся на одном диске (я полагаю, это зависит от файловой системы).

В любом случае клиенты, создающие файлы, должны будут выполнить дополнительную работу.

Answer 4

Похоже, что Apache Camel решает проблему загрузки файла, не пытаясь переименовать файл (java.io.File.renameTo).Если переименовать не удалось, нет блокировки чтения, но продолжайте попытки.Когда переименование завершается успешно, они переименовывают его обратно, а затем приступают к намеченной обработке.

См. operations.renameFile ниже.Вот ссылки на источник Apache Camel: GenericFileRenameExclusiveReadLockStrategy.java и FileUtil.java

public boolean acquireExclusiveReadLock( ... ) throws Exception {
   LOG.trace("Waiting for exclusive read lock to file: {}", file);

   // the trick is to try to rename the file, if we can rename then we have exclusive read
   // since its a Generic file we cannot use java.nio to get a RW lock
   String newName = file.getFileName() + ".camelExclusiveReadLock";

   // make a copy as result and change its file name
   GenericFile<T> newFile = file.copyFrom(file);
   newFile.changeFileName(newName);
   StopWatch watch = new StopWatch();

   boolean exclusive = false;
   while (!exclusive) {
        // timeout check
        if (timeout > 0) {
            long delta = watch.taken();
            if (delta > timeout) {
                CamelLogger.log(LOG, readLockLoggingLevel,
                        "Cannot acquire read lock within " + timeout + " millis. Will skip the file: " + file);
                // we could not get the lock within the timeout period, so return false
                return false;
            }
        }

        exclusive = operations.renameFile(file.getAbsoluteFilePath(), newFile.getAbsoluteFilePath());
        if (exclusive) {
            LOG.trace("Acquired exclusive read lock to file: {}", file);
            // rename it back so we can read it
            operations.renameFile(newFile.getAbsoluteFilePath(), file.getAbsoluteFilePath());
        } else {
            boolean interrupted = sleep();
            if (interrupted) {
                // we were interrupted while sleeping, we are likely being shutdown so return false
                return false;
            }
        }
   }

   return true;
}

Answer 5

Я знаю, что это старый вопрос, но, возможно, он может кому-нибудь помочь.

У меня была такая же проблема, поэтому я сделал следующее:

if (kind == ENTRY_CREATE) {
            System.out.println("Creating file: " + child);

            boolean isGrowing = false;
            Long initialWeight = new Long(0);
            Long finalWeight = new Long(0);

            do {
                initialWeight = child.toFile().length();
                Thread.sleep(1000);
                finalWeight = child.toFile().length();
                isGrowing = initialWeight < finalWeight;

            } while(isGrowing);

            System.out.println("Finished creating file!");

        }

Когда файл находитсясоздано, оно будет становиться все больше и больше.Поэтому я сравнил вес с точностью до секунды.Приложение будет в цикле, пока оба веса не будут одинаковыми.

Answer 6

Хотя API-интерфейс Watcher Service не может уведомить об этом, когда SO завершит копирование, все параметры кажутся «обходными» (включая этот!).

Как указано выше,

1) Перемещение или копирование не поддерживается в UNIX;

2) File.canWrite всегда возвращает true, если у вас есть разрешение на запись, даже если файл все еще копируется;

3) Ожидание, пока не произойдет тайм-аут или новое событие, но что, если система перегружена, но копирование не было завершено?если таймаут является большим значением, программа будет ждать так долго.

4) Запись другого файла в 'flag', что копирование завершено, не вариант, если вы просто используете файл, а не создаете.

Альтернативой является использование кода ниже:

boolean locked = true;

while (locked) {
    RandomAccessFile raf = null;
    try {
            raf = new RandomAccessFile(file, "r"); // it will throw FileNotFoundException. It's not needed to use 'rw' because if the file is delete while copying, 'w' option will create an empty file.
            raf.seek(file.length()); // just to make sure everything was copied, goes to the last byte
            locked = false;
        } catch (IOException e) {
            locked = file.exists();
            if (locked) {
                System.out.println("File locked: '" + file.getAbsolutePath() + "'");
                Thread.sleep(1000); // waits some time
            } else { 
                System.out.println("File was deleted while copying: '" + file.getAbsolutePath() + "'");
            }
    } finally {
            if (raf!=null) {
                raf.close();    
            }
        }
}

Answer 7

Итак, у меня была та же проблема, и у меня было следующее решение.Ранее неудачная попытка - пытался отслеживать статистику «lastModifiedTime» каждого файла, но я заметил, что увеличение размера большого файла может приостановиться на некоторое время. (Размер не изменяется непрерывно)

Основная идея - Для каждого события создайте файл триггера (во временном каталоге), имя которого имеет следующий формат -

OriginalFileName_lastModifiedTime_numberOfTries

Этот файл пуст и всеигра только во имя.Исходный файл будет рассматриваться только после прохождения интервалов определенной длительности без изменения его статистики «Время последнего изменения».(Примечание - поскольку это статистика файла, никаких накладных расходов нет -> O (1))

ПРИМЕЧАНИЕ - Этот файл триггера обрабатывается другой службой (скажем, ' FileTrigger ').

Преимущество -

1. Нет сна или ожидания для удержания системы.
2. Сбрасываетсредство просмотра файлов для мониторинга других событий

CODE для FileWatcher -

val triggerFileName: String = triggerFileTempDir + orifinalFileName + "_" + Files.getLastModifiedTime(Paths.get(event.getFile.getName.getPath)).toMillis + "_0"

// creates trigger file in temporary directory
val triggerFile: File = new File(triggerFileName)
val isCreated: Boolean = triggerFile.createNewFile()

if (isCreated)
    println("Trigger created: " + triggerFileName)
else
    println("Error in creating trigger file: " + triggerFileName)

CODE для FileTrigger (задание cron за интервал, скажем, 5 минут) -

 val actualPath : String = "Original file directory here"
 val tempPath : String = "Trigger file directory here"
 val folder : File = new File(tempPath)    
 val listOfFiles = folder.listFiles()

for (i <- listOfFiles)
{

    // ActualFileName_LastModifiedTime_NumberOfTries
    val triggerFileName: String = i.getName
    val triggerFilePath: String = i.toString

    // extracting file info from trigger file name
    val fileInfo: Array[String] = triggerFileName.split("_", 3)
    // 0 -> Original file name, 1 -> last modified time, 2 -> number of tries

    val actualFileName: String = fileInfo(0)
    val actualFilePath: String = actualPath + actualFileName
    val modifiedTime: Long = fileInfo(1).toLong
    val numberOfTries: Int = fileStats(2).toInt

    val currentModifiedTime: Long = Files.getLastModifiedTime(Paths.get(actualFilePath)).toMillis
    val differenceInModifiedTimes: Long = currentModifiedTime - modifiedTime
    // checks if file has been copied completely(4 intervals of 5 mins each with no modification)
    if (differenceInModifiedTimes == 0 && numberOfTries == 3)
    {
        FileUtils.deleteQuietly(new File(triggerFilePath))
        println("Trigger file deleted. Original file completed : " + actualFilePath)
    }
    else
    {
        var newTriggerFileName: String = null
        if (differenceInModifiedTimes == 0)
        {
            // updates numberOfTries by 1
            newTriggerFileName = actualFileName + "_" + modifiedTime + "_" + (numberOfTries + 1)
        }
        else
        {
            // updates modified timestamp and resets numberOfTries to 0
            newTriggerFileName = actualFileName + "_" + currentModifiedTime + "_" + 0
        }

        // renames trigger file
        new File(triggerFilePath).renameTo(new File(tempPath + newTriggerFileName))
        println("Trigger file renamed: " + triggerFileName + " -> " + newTriggerFileName)
    }    
}

Answer 8

Если у вас нет контроля над процессом записи, запишите все события ENTRY_CREATED и посмотрите, есть ли шаблоны .

В моем случае файлы создаются через WebDav (Apache), и создается много временных файлов, но также запускаются два события ENTRY_CREATED для одного и того же файла.Второе событие ENTRY_CREATED указывает, что процесс копирования завершен.

Вот мой пример ENTRY_CREATED событий.Печатается абсолютный путь к файлу (ваш журнал может отличаться в зависимости от приложения, которое записывает файл):

[info] application - /var/www/webdav/.davfs.tmp39dee1 was created
[info] application - /var/www/webdav/document.docx was created
[info] application - /var/www/webdav/.davfs.tmp054fe9 was created
[info] application - /var/www/webdav/document.docx was created
[info] application - /var/www/webdav/.DAV/__db.document.docx was created 

Как видите, я получаю два события ENTRY_CREATED для document.docx .После второго события я знаю, что файл готов.Временные файлы, очевидно, игнорируются в моем случае.

Answer 9

Для больших файлов в linux файлы копируются с расширением .filepart.Вам просто нужно проверить расширение с помощью api commons и зарегистрировать событие ENTRY_CREATE.Я проверил это с моими CSV-файлами (1 ГБ) и добавил, что это работает

public void run()
{
    try
    {
        WatchKey key = myWatcher.take();
        while (key != null)
        {
            for (WatchEvent event : key.pollEvents())
            {
                if (FilenameUtils.isExtension(event.context().toString(), "filepart"))
                {
                    System.out.println("Inside the PartFile " + event.context().toString());
                } else
                {
                    System.out.println("Full file Copied " + event.context().toString());
                    //Do what ever you want to do with this files.
                }
            }
            key.reset();
            key = myWatcher.take();
        }
    } catch (InterruptedException e)
    {
        e.printStackTrace();
    }
}

Answer 10

В зависимости от того, насколько срочно вам нужно переместить файл после завершения записи, вы также можете проверить стабильную последнюю измененную временную метку и переместить файл только в том случае, если он находится в режиме ожидания. Время, в течение которого он должен быть стабильным, может зависеть от реализации, но я бы предположил, что что-то с последней измененной временной меткой, которая не изменялась в течение 15 секунд, должно быть достаточно стабильным для перемещения.