Java 8 lambda api

Question

我正在努力从Rx Java迁移到Java 8 lambdas.我找不到的一个例子是缓冲请求的方法.例如,在Rx Java中,我可以说以下内容.

Observable.create(getIterator()).buffer(20, 1000, TimeUnit. MILLISECONDS).doOnNext(list -> doWrite(list));

我们将20个元素缓冲到一个列表中,或者在1000毫秒时超时,这是首先发生的.

RX中的Observable是一种"推"式可观察,而Streams使用java pull.这是否有可能在流中实现我自己的map操作,或者由于doOnNext必须轮询前一个元素而导致无法发出导致问题？

Answer 1

一种方法是使用 BlockingQueue 和 Guava。使用Queues.drain，您可以创建一个Collection然后可以调用stream()并进行转换的 。这是一个链接：Guava Queues.drain

这是一个简单的例子：

public void transform(BlockingQueue<Something> input) 
{
    List<Something> buffer = new ArrayList<>(20);
    Queues.drain(input, buffer, 20, 1000, TimeUnit.MILLISECONDS);
    doWrite(buffer);
}