如何从阻塞队列中创建Reactor通量？

Question

我试图实现从阻塞队列创建的Reactor通量，但不确定哪个操作符最适合我的用例？

我正在创建一个流式RESTendpoint，其中的响应是流量，需要不断从阻塞队列中发出消息，作为获取REST调用的响应。

我已经尝试过论坛和留档，只能找到从可迭代集合或响应数据源发起的Flux，但没有任何BlockingQueue的示例。

Answer 1

你可以试试Flux#generate and Queue#peek。请记住，如果队列为空，peek将返回null，并且不能在onNext中使用。

比如：

Flux.generate(sink -> {
    val element = queue.peek();
    if (element == null) {
        sink.complete();
    } else {
        sink.next(element);
    }
});

还有Flux#repeatWhen操作符，以防在队列被认为是空的情况下重新订阅队列，例如：

flux.repeatWhen(it -> it.delayElements(ofSeconds(1)))