Stream API и очереди: подписка на BlockingQueue в стиле потока

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

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

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

Stream.generate(() -> {
        try {
            return queue.take();
        } catch (InterruptedException ie) {
            return "Interrupted!";
        }
    })
    .filter(s -> s.endsWith("x"))
    .forEach(System.out::println);   

(К сожалению, необходимость обрабатывать InterruptedException делает это довольно запутанным.)

Обратите внимание, что нет способа закрыть очередь, и Stream.generate не может остановить создание элементов, так что это фактически бесконечный поток. Единственный способ завершить его — использовать операцию короткого замыкания потока, такую ​​как findFirst.

Вы можете посмотреть на реализацию асинхронной очереди. Если у вас Java 8, то cyclops-react, я разработчик этого проекта, предоставляет асинхронный .Queue, который позволит вам как заполнять, так и потреблять из очереди асинхронно (и чисто).

e.g.

Queue<String> queue = QueueFactories.<String>unboundedQueue().build();

Или просто (если это com.aol.simple.react.async.Queue)

Queue<String> queue = new Queue<>();

Потом в отдельной теме:

new Thread(() -> {
        while (true) {
            queue.add("New message " + System.currentTimeMillis());
        }
    }).start();

Вернувшись в основной поток, исходный код теперь должен работать так, как ожидалось (бесконечно перебирать сообщения, добавляемые в очередь, и распечатывать их).

queue.stream().forEach(e -> System.out.println(e));

Очередь и, следовательно, поток могут быть закрыты на любом этапе через -

queue.close();

Другой подход заключается в создании собственного Spliterator. В моем случае у меня есть очередь блокировки, и я хочу создать поток, который продолжает извлекать элементы, пока блок не истечет. Сплитератор примерно такой:

public class QueueSpliterator<T> implements Spliterator<T> {
    private final BlockingQueue<T> queue;
    private final long timeoutMs;

    public QueueSpliterator(final BlockingQueue<T> queue, final long timeoutMs) {
        this.queue = queue;
        this.timeoutMs = timeoutMs;
    }

    @Override
    public int characteristics() {
        return Spliterator.CONCURRENT | Spliterator.NONNULL | Spliterator.ORDERED;
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE;
    }

    @Override
    public boolean tryAdvance(final Consumer<? super T> action) {
        try {
            final T next = this.queue.poll(this.timeoutMs, TimeUnit.MILLISECONDS);
            if (next == null) {
                return false;
            }
            action.accept(next);
            return true;
        } catch (final InterruptedException e) {
            throw new SupplierErrorException("interrupted", e);
        }
    }

    @Override
    public Spliterator<T> trySplit() {
        return null;
    }

}

Исключение, созданное для обработки InterruptedException, является расширением RuntimeException. Используя этот класс, можно построить поток через: StreamSupport.stream(new QueueSpliterator(...)) и добавить обычные операции потока.