当一定超时后没有传入数据时终止流

Question

我有一个InputStreamand OutputStream（没有套接字）。

我有一个基于流的代码，可以执行一些映射/过滤/分组/处理。

我的主要目标是在超出时终止流maxDuration：

void fillStreamMap(BufferedReader reader) {
    final Instant end = Instant.now().plusNanos(TimeUnit.NANOSECONDS.convert(maxDuration));

    this.map = reader.lines()
        .takeWhile(e -> checkTimeout(end))
        .map(this::jsonToBuyerEventInput)
        .filter(Objects::nonNull)
        .filter(getFilter()::apply)
        .limit(super.maxEvent)
        .collect(Collectors.groupingBy(BuyerEventInput::getBuyer));
}

boolean checkTimeout(Instant end){
    return Instant.now().getEpochSecond() <= end.getEpochSecond();
}

我正在使用takeWhile这是一个非常有用的函数，但它会检查终止条件是否有即将发生的事件。

因此，如果没有发送数据，它不会检查条件，因为该函数是为了将 aPredicate作为参数而构建的。

有什么办法可以实现这个目标吗？

Answer 1

这是一种在 Streams 上运行的方法。核心功能是timedTake(Stream<T> stream, long timeout, TimeUnit unit). 这个想法是使用其原始Spliterator遍历原始流，这使得设置超时成为可能。

\n

import java.io.BufferedReader;\nimport java.io.InputStreamReader;\nimport java.util.Optional;\nimport java.util.Spliterator;\nimport java.util.concurrent.ExecutionException;\nimport java.util.concurrent.ExecutorService;\nimport java.util.concurrent.Executors;\nimport java.util.concurrent.Future;\nimport java.util.concurrent.ThreadFactory;\nimport java.util.concurrent.TimeUnit;\nimport java.util.concurrent.TimeoutException;\nimport java.util.function.Supplier;\nimport java.util.stream.Stream;\n\nclass Main {\n    static <T> Stream<T> generateOrderedStream(Supplier<Optional<T>> s) {\n        // Returns an ordered stream with the values of the Optionals returned by s.get(). An empty Optional ends the stream.\n        // As pseudocode:\n        //     for (Optional<T> o = s.get(); o.isPresent(); o = s.get())\n        //         emit o.get();\n        return Stream.iterate(s.get(), Optional::isPresent, prev -> s.get())\n            .map(Optional::get);\n    }\n\n    static <T> Optional<T> advance(Spliterator<T> iter) {\n        // Returns an Optional with the next element of the iterator, or an empty Optional if there are no more elements.\n        // (This method is much nicer than calling iter.tryAdvance() directly.)\n        final var r = new Object() { T elem; };\n        return iter.tryAdvance(elem -> r.elem = elem) ? Optional.of(r.elem) : Optional.empty();\n    }\n\n    static ThreadFactory daemonThreadFactory() {\n        return (r) -> {\n            Thread thread = new Thread(r);\n            thread.setDaemon(true);\n            return thread;\n        };\n    }\n\n    static <T> Stream<T> timedTake(Stream<T> stream, long timeout, TimeUnit unit) {\n        // Traverses the stream until the timeout elapses and returns the traversed elements.\n        final long deadlineNanos = System.nanoTime() + unit.toNanos(timeout);\n        final ExecutorService executor = Executors.newSingleThreadExecutor(daemonThreadFactory());\n        final Spliterator<T> iter = stream.spliterator();\n        return generateOrderedStream(() -> {\n            try {\n                Future<Optional<T>> future = executor.submit(() -> advance(iter));\n                long remainingNanos = deadlineNanos - System.nanoTime();\n                Optional<T> optElem = future.get(remainingNanos, TimeUnit.NANOSECONDS);\n                if (!optElem.isPresent()) { // this is the end of the input stream, so clean up\n                    executor.shutdownNow();\n                }\n                return optElem;\n            } catch (TimeoutException e) {\n                executor.shutdownNow();\n                return Optional.empty(); // mark this as the end of the result stream\n            } catch (ExecutionException e) {\n                executor.shutdownNow();\n                throw new RuntimeException(e.getCause());\n            } catch (InterruptedException e) {\n                executor.shutdownNow();\n                throw new RuntimeException(e);\n            }\n        });\n    }\n\n    static void fillStreamMap(BufferedReader reader) {\n        // streaming demo\n        long maxDurationSecs = 5;\n        timedTake(reader.lines(), maxDurationSecs, TimeUnit.SECONDS)\n            .takeWhile(line -> !line.contains("[stop]"))\n            .map(line -> "[mapped] " + line)\n            .forEachOrdered(System.out::println);\n    }\n\n    public static void main(String[] args) {\n        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));\n        fillStreamMap(reader);\n    }\n}\n

\n\n

另一种方法是在 Reader 级别进行操作，并从 中读取超时BufferedReader（可能会换行System.in）。不幸的是，很难正确地做到这一点（例如，请参阅为用户输入设置超时，以及控制台输入超时一文）。

\n

这些链接页面的一个想法是轮询BufferedReader.ready()直到它返回true，然后调用readLine()。这是丑陋的（因为它使用轮询）并且不可靠，因为readLine()即使ready()返回 true \xe2\x80\x93 也可能会阻塞，例如因为有不完整的行可用（在类 Unix 系统上，用户可以通过键入一些文本然后按Ctrl+D 而不是 Enter）。

\n

另一个想法是创建一个后台线程，重复调用BufferedReader.readLine()并将结果插入到BlockingQueue（例如ArrayBlockingQueue）中。然后主线程可以在队列上调用take()或poll(timeout,unit)来获取行。

\n

这种方法的一个限制是，如果您稍后想要直接读取BufferedReader（而不是通过队列），则几乎不可能避免丢失（至少）一行输入。这是因为线程在被阻塞时无法被干净地中断readLine()，因此如果主线程决定提前停止（例如由于超时），它无法阻止后台线程读取它所在的行目前正在等待。

\n

您可以尝试使用mark(readAheadLimit)和reset()BufferedReader “取消读取”最后一行，但同步会很困难 \xe2\x80\x93 另一个线程可能会尝试在后台线程调用之前读取reset()。您可能必须使用锁定字段进行同步，但是它的访问级别是protected这样的，因此您只能使用反射或通过子类化来访问它BufferedReader。另外，reset()如果未读行的长度超过 ，则会失败readAheadLimit。

\n

这是一个假设您仅通过队列读取行的实现。

\n

免责声明：请注意这些代码片段中的错误 \xe2\x80\x93 多线程很棘手。我可能会再次尝试改进代码。

\n

import java.io.BufferedReader;\nimport java.io.IOException;\nimport java.io.InputStreamReader;\nimport java.util.Optional;\nimport java.util.concurrent.ArrayBlockingQueue;\nimport java.util.concurrent.TimeUnit;\nimport java.util.function.Supplier;\nimport java.util.stream.Stream;\n\nclass InterruptibleLineReader {\n    private static final String EOF = new String("<EOF>");\n    BufferedReader reader;\n    ArrayBlockingQueue<String> lines = new ArrayBlockingQueue<>(/* capacity: */ 2);\n    Thread backgroundThread;\n    IOException exception;\n\n    public InterruptibleLineReader(BufferedReader reader) {\n        this.reader = reader;\n        // start a background thread to read lines\n        backgroundThread = new Thread(this::backgroundTask);\n        backgroundThread.setDaemon(true);\n        backgroundThread.start();\n    }\n\n    public void close() {\n        backgroundThread.interrupt();\n        lines.clear();\n        lines.add(EOF);\n    }\n\n    private void backgroundTask() {\n        try {\n            try {\n                while (true) {\n                    String line = reader.readLine();\n                    if (Thread.interrupted()) {\n                        // nothing to do (close() is responsible for lines.put(EOF) etc. in this case)\n                        break;\n                    } else if (line == null) {\n                        lines.put(EOF);\n                        break;\n                    }\n                    lines.put(line);\n                }\n            } catch (IOException e) {\n                exception = e;\n                lines.put(EOF);\n            }\n        } catch (InterruptedException e) {\n            // nothing to do (close() is responsible for lines.put(EOF) etc. in this case)\n        }\n    }\n\n    public String readLine(long timeout, TimeUnit unit) throws IOException, InterruptedException {\n        String line = lines.poll(timeout, unit);\n        if (line == EOF) { // EOF or IOException\n            lines.put(EOF); // restore the EOF so that any concurrent (and future) calls to this method won\'t block\n            if (exception != null) {\n                throw exception;\n            } else {\n                return null;\n            }\n        }\n        return line;\n    }\n}\n\nclass Main {\n    static <T> Stream<T> generateOrderedStream(Supplier<Optional<T>> s) {\n        // Returns an ordered stream with the values of the Optionals returned by s.get(). An empty Optional ends the stream.\n        // As pseudocode:\n        //     for (Optional<T> o = s.get(); o.isPresent(); o = s.get())\n        //         emit o.get();\n        return Stream.iterate(s.get(), Optional::isPresent, prev -> s.get())\n            .map(Optional::get);\n    }\n\n    static Stream<String> timedReadLines(InterruptibleLineReader lineReader, long timeout, TimeUnit unit) {\n        // Reads lines until the timeout elapses and returns them as a stream.\n        final long deadlineNanos = System.nanoTime() + unit.toNanos(timeout);\n        return generateOrderedStream(() -> {\n            try {\n                long remainingNanos = deadlineNanos - System.nanoTime();\n                return Optional.ofNullable(lineReader.readLine(remainingNanos, TimeUnit.NANOSECONDS));\n            } catch (IOException|InterruptedException e) {\n                throw new RuntimeException(e);\n            }\n        });\n    }\n\n    static void fillStreamMap(InterruptibleLineReader lineReader) {\n        // streaming demo\n        long maxDurationSecs = 5;\n        timedReadLines(lineReader, maxDurationSecs, TimeUnit.SECONDS)\n            .takeWhile(line -> !line.contains("[stop]"))\n            .map(line -> "[mapped] " + line)\n            .forEachOrdered(System.out::println);\n    }\n\n    public static void main(String[] args) {\n        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));\n\n        // stream lines\n        InterruptibleLineReader lineReader = new InterruptibleLineReader(reader);\n        fillStreamMap(lineReader);\n        lineReader.close();\n\n        /*\n        // attempt to use the BufferedReader directly\n        // NOTE: several lines may be lost (depending on the capacity of the ArrayBlockingQueue and how quickly the lines are consumed)\n        System.out.println("--- reading directly from BufferedReader ---");\n        while (true) {\n            try {\n                String line = reader.readLine();\n                if (line == null) { break; }\n                System.out.println("[raw] " + line);\n            } catch (IOException e) {\n                throw new RuntimeException(e);\n            }\n        }\n        */\n    }\n}\n

\n

这是一种更复杂的实现，如果关闭队列并直接从 .txt 文件中读取，则只会丢失一行输入BufferedReader。它使用自定义的“0容量”队列来确保最多丢失一行。

\n

import java.io.BufferedReader;\nimport java.io.IOException;\nimport java.io.InputStreamReader;\nimport java.util.NoSuchElementException;\nimport java.util.Optional;\nimport java.util.concurrent.TimeUnit;\nimport java.util.function.Supplier;\nimport java.util.stream.Stream;\n\nclass InterruptibleLineReader {\n    BufferedReader reader;\n    ZeroCapacityBlockingQueue<String> lines = new ZeroCapacityBlockingQueue<>(); // a null line indicates EOF or IOException\n    Thread backgroundThread;\n    IOException exception;\n    boolean eof;\n\n    public InterruptibleLineReader(BufferedReader reader) {\n        this.reader = reader;\n        // start a background thread to read lines\n        backgroundThread = new Thread(this::backgroundTask);\n        backgroundThread.setDaemon(true);\n        backgroundThread.start();\n    }\n\n    private void markAsEOF() {\n        eof = true;\n        if (lines.poll() != null) { // markAsEOF() should not be called when there are unconsumed lines\n            throw new IllegalStateException();\n        }\n        lines.offer(null); // unblock threads that are waiting on the queue\n    }\n\n    public void close() {\n        backgroundThread.interrupt();\n        // warn if there is an unconsumed line, and consume it so we can indicate EOF\n        String line = lines.poll();\n        if (line != null) {\n            System.err.println("InterruptibleLineReader: warning: discarding unconsumed line during close(): \'" + line + "\'");\n        }\n        markAsEOF();\n    }\n\n    private void backgroundTask() {\n        try {\n            while (true) {\n                String line = reader.readLine();\n                if (Thread.interrupted()) {\n                    if (line != null) {\n                        System.err.println("InterruptibleLineReader: warning: discarding line that was read after close(): \'" + line + "\'");\n                    }\n                    // nothing further to do (close() is responsible for calling markAsEOF() in this case)\n                    break;\n                } else if (line == null) { // EOF\n                    markAsEOF();\n                    break;\n                }\n                lines.put(line); // this blocks until the line has been consumed ("0-capacity" behaviour)\n                if (Thread.interrupted()) {\n                    // nothing to do (close() is responsible for calling markAsEOF() in this case)\n                    break;\n                }\n            }\n        } catch (IOException e) {\n            exception = e;\n            markAsEOF();\n        } catch (InterruptedException e) {\n            // nothing to do (close() is responsible for calling markAsEOF() in this case)\n        }\n    }\n\n    public String readLine() throws IOException, InterruptedException {\n        String line = lines.take();\n        if (line == null) { // EOF or IOException\n            markAsEOF(); // restore the null so that any concurrent (and future) calls to this method won\'t block\n            if (exception != null) {\n                throw exception;\n            } else {\n                return null; // EOF\n            }\n        } else {\n            return line;\n        }\n    }\n\n    public String readLine(long timeout, TimeUnit unit) throws IOException, InterruptedException {\n        String line = lines.poll(timeout, unit);\n        if (line == null && eof) { // EOF or IOException (not timeout)\n            markAsEOF(); // restore the null so that any concurrent (and future) calls to this method won\'t block\n            if (exception != null) {\n                throw exception;\n            } else {\n                return null; // EOF\n            }\n        } else {\n            return line;\n        }\n    }\n}\n\nclass ZeroCapacityBlockingQueue<T> {\n    int count;\n    T item;\n\n    public synchronized boolean add(T x) {\n        // does not block (i.e. behaves as if the capacity is actually 1)\n        if (count == 1) {\n            throw new IllegalStateException("Queue full");\n        }\n        item = x;\n        count++;\n        notifyAll();\n        return true;\n    }\n\n    public synchronized boolean offer(T x) {\n        // does not block (i.e. behaves as if the capacity is actually 1)\n        if (count == 1) {\n            return false;\n        }\n        return add(x);\n    }\n\n    public synchronized void put(T x) throws InterruptedException {\n        // blocks until the item has been removed ("0-capacity" behaviour)\n        while (count == 1) {\n            wait();\n        }\n        add(x);\n        while (count == 1 && item == x) {\n            wait();\n        }\n    }\n\n    public synchronized T remove() {\n        if (count == 0) {\n            throw new NoSuchElementException();\n        }\n        T x = item;\n        item = null;\n        count--;\n        notifyAll();\n        return x;\n    }\n\n    public synchronized T poll() {\n        if (count == 0) {\n            return null;\n        }\n        return remove();\n    }\n\n    public synchronized T take() throws InterruptedException {\n        while (count == 0) {\n            wait();\n        }\n        return remove();\n    }\n\n    public synchronized T poll(long timeout, TimeUnit unit) throws InterruptedException {\n        long deadlineNanos = System.nanoTime() + unit.toNanos(timeout);\n        while (count == 0) {\n            long remainingNanos = deadlineNanos - System.nanoTime();\n            if (remainingNanos <= 0) {\n                return null;\n            }\n            TimeUnit.NANOSECONDS.timedWait(this, remainingNanos);\n        }\n        return remove();\n    }\n}\n\nclass Main {\n    static <T> Stream<T> generateOrderedStream(Supplier<Optional<T>> s) {\n        // Returns an ordered stream with the values of the Optionals returned by s.get(). An empty Optional ends the stream.\n        // As pseudocode:\n        //     for (Optional<T> o = s.get(); o.isPresent(); o = s.get())\n        //         emit o.get();\n        return Stream.iterate(s.get(), Optional::isPresent, prev -> s.get())\n            .map(Optional::get);\n    }\n\n    static Stream<String> timedReadLines(InterruptibleLineReader lineReader, long timeout, TimeUnit unit) {\n        // Reads lines until the timeout elapses and returns them as a stream.\n        final long deadlineNanos = System.nanoTime() + unit.toNanos(timeout);\n        return generateOrderedStream(() -> {\n            try {\n                long remainingNanos = deadlineNanos - System.nanoTime();\n                return Optional.ofNullable(lineReader.readLine(remainingNanos, TimeUnit.NANOSECONDS));\n            } catch (IOException|InterruptedException e) {\n                throw new RuntimeException(e);\n            }\n        });\n    }\n\n    static void fillStreamMap(InterruptibleLineReader lineReader) {\n        // streaming demo\n        long maxDurationSecs = 5;\n        timedReadLines(lineReader, maxDurationSecs, TimeUnit.SECONDS)\n            .takeWhile(line -> !line.contains("[stop]"))\n            .map(line -> "[mapped] " + line)\n            .forEachOrdered(System.out::println);\n    }\n\n    public static void main(String[] args) {\n        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));\n\n        // stream lines\n        InterruptibleLineReader lineReader = new InterruptibleLineReader(reader);\n        fillStreamMap(lineReader);\n        lineReader.close();\n\n        /*\n        // attempt to use the BufferedReader directly\n        // NOTE: a line will be lost\n        System.out.println("--- reading directly from BufferedReader ---");\n        while (true) {\n            try {\n                String line = reader.readLine();\n                if (line == null) { break; }\n                System.out.println("[raw] " + line);\n            } catch (IOException e) {\n                throw new RuntimeException(e);\n            }\n        }\n        */\n    }\n}\n

\n

这是第二个实现的示例运行（最后一部分main()未注释）。时间戳以秒为单位，“>”表示输入。

\n

import java.io.BufferedReader;\nimport java.io.InputStreamReader;\nimport java.util.Optional;\nimport java.util.Spliterator;\nimport java.util.concurrent.ExecutionException;\nimport java.util.concurrent.ExecutorService;\nimport java.util.concurrent.Executors;\nimport java.util.concurrent.Future;\nimport java.util.concurrent.ThreadFactory;\nimport java.util.concurrent.TimeUnit;\nimport java.util.concurrent.TimeoutException;\nimport java.util.function.Supplier;\nimport java.util.stream.Stream;\n\nclass Main {\n    static <T> Stream<T> generateOrderedStream(Supplier<Optional<T>> s) {\n        // Returns an ordered stream with the values of the Optionals returned by s.get(). An empty Optional ends the stream.\n        // As pseudocode:\n        //     for (Optional<T> o = s.get(); o.isPresent(); o = s.get())\n        //         emit o.get();\n        return Stream.iterate(s.get(), Optional::isPresent, prev -> s.get())\n            .map(Optional::get);\n    }\n\n    static <T> Optional<T> advance(Spliterator<T> iter) {\n        // Returns an Optional with the next element of the iterator, or an empty Optional if there are no more elements.\n        // (This method is much nicer than calling iter.tryAdvance() directly.)\n        final var r = new Object() { T elem; };\n        return iter.tryAdvance(elem -> r.elem = elem) ? Optional.of(r.elem) : Optional.empty();\n    }\n\n    static ThreadFactory daemonThreadFactory() {\n        return (r) -> {\n            Thread thread = new Thread(r);\n            thread.setDaemon(true);\n            return thread;\n        };\n    }\n\n    static <T> Stream<T> timedTake(Stream<T> stream, long timeout, TimeUnit unit) {\n        // Traverses the stream until the timeout elapses and returns the traversed elements.\n        final long deadlineNanos = System.nanoTime() + unit.toNanos(timeout);\n        final ExecutorService executor = Executors.newSingleThreadExecutor(daemonThreadFactory());\n        final Spliterator<T> iter = stream.spliterator();\n        return generateOrderedStream(() -> {\n            try {\n                Future<Optional<T>> future = executor.submit(() -> advance(iter));\n                long remainingNanos = deadlineNanos - System.nanoTime();\n                Optional<T> optElem = future.get(remainingNanos, TimeUnit.NANOSECONDS);\n                if (!optElem.isPresent()) { // this is the end of the input stream, so clean up\n                    executor.shutdownNow();\n                }\n                return optElem;\n            } catch (TimeoutException e) {\n                executor.shutdownNow();\n                return Optional.empty(); // mark this as the end of the result stream\n            } catch (ExecutionException e) {\n                executor.shutdownNow();\n                throw new RuntimeException(e.getCause());\n            } catch (InterruptedException e) {\n                executor.shutdownNow();\n                throw new RuntimeException(e);\n            }\n        });\n    }\n\n    static void fillStreamMap(BufferedReader reader) {\n        // streaming demo\n        long maxDurationSecs = 5;\n        timedTake(reader.lines(), maxDurationSecs, TimeUnit.SECONDS)\n            .takeWhile(line -> !line.contains("[stop]"))\n            .map(line -> "[mapped] " + line)\n            .forEachOrdered(System.out::println);\n    }\n\n    public static void main(String[] args) {\n        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));\n        fillStreamMap(reader);\n    }\n}\n

\n

注意“四”行是如何丢失的。为避免丢失行，创建实例BufferedReader后请勿使用底层。InterruptibleLineReader

\n

（如果您确实需要在BufferedReader那之后，您可以编写一个虚拟子类来BufferedR