Java 8 流和传入数据 (Kafka)

Question

我有一个队列（它恰好是 Kafka，但我不确定这是否重要），我正在从中读取消息。我想创建一个流来表示这些数据。

我使用（Kafka）队列的伪代码如下所示：

List<Message> messages = new ArrayList<>();

while (true) {
    ConsumerRecords<String, Message> records = kafkaConsumer.poll(100);

    messages.add(recordsToMessages(records));

    if (x) {
        break;
    }
}

return messages.stream();

使用此伪代码，直到 while 循环被破坏，即直到所有队列都被读取后，才会返回流。

我希望能够立即返回流，即可以将新消息添加到流在返回之后。

我觉得我需要使用 Stream.generate 但我不确定如何使用，或者我需要一个拆分器？

我还想稍后在代码中关闭流。

谢谢！

Answer 1

这是一个如何完成的注释示例：

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();
    }
}

这段代码背后的想法是，您可以通过自定义Iterator 提供Stream，它本身从外部来源提取数据。

数据通过Queue 从外部源传输到Iterator。并且因为只有您知道您的数据是什么样子以及如何检测没有任何剩余可读取的，所以确定是否应继续提供 Stream 的过程被提取到用户提供的函数中。

希望有帮助吗？