下面是一个注解示例，说明了如何进行此操作：

public static void main(String[] args) {

    LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

    // Data producer
    Runnable job = () -> {
        // Send data to the stream (could come from your Kafka queue
        ThreadLocalRandom random = ThreadLocalRandom.current();
        for (int i = 0; i < 10; i++) {
            queue.offer(random.nextInt(100));
            delay(random.nextInt(2) + 1);
        }
        // Send the magic signal to stop the stream
        queue.offer(-1);
    };
    Thread thread = new Thread(job);
    thread.start();

    // Define the condition by which the stream knows there is no data left to consume
    // The function returns the next element wrapped in an Optional, or an empty Optional to tell there is no more data to read
    // In this example, the number -1 is the magic signal
    Function<BlockingQueue<Integer>, Optional<Integer>> endingCondition = q -> {
        try {
            Integer element = q.take();
            return element == -1 ? Optional.empty() : Optional.of(element);
        } catch (InterruptedException e) {
            return Optional.empty();
        }
    };
    QueueConsumingIterator<Integer> iterator = new QueueConsumingIterator<>(queue, endingCondition);

    // Construct a Stream on top of our custom queue-consuming Iterator
    Spliterator<Object> spliterator = Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED);
    Stream<Object> stream = StreamSupport.stream(spliterator, false);

    // Use the Stream as usual :)
    stream.map(String::valueOf).forEach(System.out::println);

}

.

// This is a custom Iterator that takes data from a BlockingQueue.
// Detection of the end of the data stream is use-case-dependant, so it is extracted as a user-provided Function<Queue, Optional>
// For example you may want to wait for a particular item in the queue, or consider the queue "dead"" after a certain timeout...
public static class QueueConsumingIterator<E> implements Iterator<E> {

    private final BlockingQueue<E> queue;
    private final Function<BlockingQueue<E>, Optional<E>> queueReader;
    private Optional<E> element;
    private boolean elementRead = false;

    public QueueConsumingIterator(BlockingQueue<E> queue, Function<BlockingQueue<E>, Optional<E>> queueReader) {
        this.queue = queue;
        this.queueReader = queueReader;
    }

    @Override
    public boolean hasNext() {
        if (!this.elementRead) {
            this.element = this.queueReader.apply(this.queue);
            this.elementRead = true;
        }
        return this.element.isPresent();
    }

    @Override
    public E next() {
        if (hasNext()) {
            this.elementRead = false;
            return this.element.get();
        }
        throw new NoSuchElementException();
    }

}

private static void delay(int timeout) {
    try {
        TimeUnit.SECONDS.sleep(timeout);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

这个代码背后的想法是，你可以养活一个 Stream 通过习俗 Iterator ，本身从外部源提取数据。
数据从外部源传输到 Iterator 通过 Queue . 因为只有你知道你的数据是什么样子的，以及如何检测到已经没有剩余的数据可以读取了，所以决定流是否应该继续被输入的过程被提取到一个用户提供的函数中。
希望有帮助？