Потоки очереди, если предыдущие не закончились

Question

Я пытаюсь написать простой видео манипулятор, поэтому несколько раз в секунду мне нужно запустить новый поток (в настоящее время реализующий Runnable) для обработки текущего кадра, но у меня нет гарантии, сколько времени займет завершение каждого потока, и поэтому я хочу ограничить количество потоков, которые могут работать одновременно, количеством процессоров на компьютере:

Runtime runtime = Runtime.getRuntime();
int nP = runtime.availableProcessors();  

Но я должен гарантировать, что все созданные потоки выполняются в последовательности, поэтому кадры не удаляются.

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

Возможно ли это, используя любую комбинацию futureTask, Exector или ExecutorService?

Благодарю.

РЕДАКТИРОВАТЬ:

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

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

Прочитав первые 3 комментария, я вижу, что это, вероятно, менее эффективный способ сделать это, и я предполагаю, что будет работать только один поток, чтобы поддерживать отзывчивость интерфейса, но я не уверен, как продолжать добавлять изображения в поток для это обрабатывать без использования огромного списка. Я предполагаю, что это будет что-то вроде:

В основном классе:

Timer actionPerformed {
    List.add(decodedImage);
}

В управляемом классе:

run() {
   while( timer.isRunning() ) {
     if( runCount >= list.size()-1 ) {
        try {
          Thread.sleep(500);
        } catch() {
             /* Catchy stuff */
        }
     } else {
        BufferedImage toProcess = list.get(runCount);
        /* Do Processing here */
        writeImageToStream();
        list.remove(runCount);
        runCount++;
     }
   }
}

Это правильно?

РЕДАКТИРОВАТЬ 2:

Итак, это то, что я имею до сих пор:

public class timerEncode {

     private long startTime;

     ActionListener goAction = new ActionListener() {
         public void actionPerformed( ActionEvent evt ) {
             BufferedImage decoded = getNextImage();
             long write_time = System.nanoTime();
             new doImages(decoded, write_time).run();
         }        
     };
     Timer goTimer = new Timer(40,goAction);

     private BufferedImage getNextImage() {
        /* Does inconsequential stuff to retrieve image from the stream*/
     }

     private void recBtnActionPerformed(java.awt.event.ActionEvent evt) {                                       
        startTime = System.nanoTime();
        goTimer.start();
     }

     private class doImages implements Runnable {
        final BufferedImage image;
        final long write_time;

        public doImages(BufferedImage image, long write_time) {
           this.image = image;
           this.write_time = write_time;
        }

        public void run() {
            BufferedImage out = toXuggleType(image, BufferedImage.TYPE_3BYTE_BGR);
            /* Other time consuming processy stuff goes here */
            /* Encode the frame to a video stream */
            writer.encodeVideo(0,out,write_time-startTime, TimeUnit.NANOSECONDS);
        }

        private BufferedImage toType(BufferedImage source, int type) {
            if( source.getType() != type ) {
                BufferedImage temp = new BufferedImage(source.getWidth(),source.getHeight(),type);
                temp.getGraphics().drawImage(source, 0, 0, null);
                source = temp;
            }
            return source;
        }
    }

}

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

Answer 1

Но я должен гарантировать, что все созданные потоки выполняются в последовательности, поэтому кадры не удаляются.

Вы имеете в виду это, как вы сформулировали это здесь? Если это так, то вы вообще не можете многопоточность, так как я понимаю, что вы не можете начать обработку кадра 2, пока кадр 1 не закончится. В этот момент вы также можете последовательно обрабатывать кадры и игнорировать многопоточность.

В качестве альтернативы, если вы имеете в виду что-то еще, например, то, что кадры могут обрабатываться независимо, но должны быть сопоставлены последовательно, то это может быть выполнимо.

В любом случае - использование "сырых" потоков редко требуется или выгодно. Как уже отмечали другие, используйте высокоуровневые утилиты параллелизма (в этом случае ThreadPoolExecutor будет идеальным) для наблюдения за этим.

Это звучит как Runnable Это не совсем правильный выбор, поскольку это означает, что вы возвращаете "результат" обработки, изменяя некоторую глобальную переменную. Вместо этого может быть лучше превратить эту обработку в Callable и вернуть результат. Это, вероятно, устранит проблемы с безопасностью потоков, может позволить обрабатывать разные кадры одновременно с меньшими проблемами и позволит отложить сопоставление каждого результата до любой точки, которую вы считаете подходящей.

Если вы хотите пойти по этому пути, вы можете сделать что-то вроде следующего:

// Create a thread pool with the given concurrency level
ExecutorService executor = Executors.newFixedThreadPool(Runtime.availableProcessors);

// Submit all tasks to the pool, storing the futures for further reference
// The ? here should be the class of object returned by your Callables
List<Future<?>> futures = new ArrayList<Future<?>>(NUM_FRAMES);
for (int i = 0; i < NUM_FRAMES; i++)
{
    futures.add(executor.submit(createCallableForFrame(i)));
}

// Combine results using future.get()
// e.g. do something with frames 2 and 3:
mergeFrames(futures.get(2).get(), futures.get(3).get());

// In practice you'd probably iterate through the Futures but it's your call!

Answer 2

Это плохая идея - запускать потоки непрерывно - это большой удар по производительности. Вам нужен пул потоков и куча заданий (Runnables). Если вы создадите пул потоков размером = число процессоров и просто продолжаете добавлять кадры (в виде заданий) в очередь заданий, ваши потоки смогут эффективно обрабатывать свой путь в очереди по порядку.

Answer 3

[смотрите историю предыдущих ответов]

Я вижу, что происходит сейчас. Я не совсем уверен, как Timer или же ActionListeners работать (в отношении того, что происходит, если предыдущий вызов не закончился, когда приходит другой), но, похоже, вы на самом деле не выполняете свой doImages объекты одновременно - запустить Runnable объект одновременно вам нужно сделать Thread t = new Thread(runnableObject); t.start(); Если вы просто позвоните run() метод, это будет сделано последовательно (как и любой другой вызов метода), так что ваш actionPerformed() метод не закончится до run() делает. Я не уверен, что это помешает (или задержит) других ActionEvents обрабатывается.

Как было предложено другими, чтобы ограничить количество потоков, вы должны использовать ThreadPoolExcecutor объект. Это сделает ваш actionPerformed() метод вернуться быстро, запустить doImages объекты одновременно и убедитесь, что вы не используете слишком много потоков через очередь. Все, что вам нужно сделать, это заменить new doImages(decoded, write_time).run(); с threadPool.execute(new doImages(decoded, write_time)),

Что касается того, как контролировать процесс, вы можете использовать getQueue() метод ThreadPoolExcecutor получить и проверить размер очереди, чтобы увидеть, сколько кадров ожидает обработки.