Java 8 Streams - Timeout?

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

ех.

final long start = System.currentTimeMillis();
myDataStructure.stream()
    .while(() -> System.currentTimeMillis() <= start + 30000)
    .forEach(e ->
    {
      ...
    });

Я хочу, чтобы во время вызова forEach просто сказать return, если выполняется определенное условие.

Ответы

Ответ 1

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

final long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(30L);
myDataStructure.stream()
    .filter(e -> System.nanoTime() <= end)
    .forEach(e ->
    {
      ...
    });

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

Ответ 2

Я бы создал для этого настраиваемый пул, например:

ForkJoinPool forkJoinPool = new ForkJoinPool(1);
    try {
        forkJoinPool.submit(() ->
        IntStream.range(1, 1_000_000).filter(x -> x > 2).boxed().collect(Collectors.toList()))
                .get(30, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        // job not done in your interval
    }

Ответ 3

Поскольку Stream forEach не имеет break, я думаю, вы можете создать Custom Exception для этого в break цикл:

myDataStructure.stream()
    .forEach(e ->
    {
      if (System.currentTimeMillis() <= start + 30000) {
          throw new MyTimeOutException()
      }
    });

и вы можете поймать этот Исключение для catch.

Ответ 4

Вы можете использовать тот факт, что .allMatch() является оператором короткого замыкания для завершения потока:

final long start = System.currentTimeMillis();
myDataStructure.stream()
    .allMatch(e ->
    {
      // your task here
        return System.currentTimeMillis() <= start + 30000;
    });

Ответ 5

Как говорится в комментариях в OP, takeWhile/dropWhile пропущены в Java 8 (будет добавлен в Java 9). Нет никаких оснований пытаться реализовать логику с помощью исключений или других кодов, потому что код просто выглядит настолько уродливым и нецензурным, даже для практики. Я думаю, что использование сторонней библиотеки - гораздо лучшее решение, например StreamEx

StreamEx(source).takeWhile(() -> System.currentTimeMillis() <= start + 30000)
                .forEach(e -> { ... });