假設(shè)我有一個(gè)具有以下簽名的函數(shù)：class Item { String name; Long id;}public Flux<Item> getNew(long id);getNew()返回在 id (0..N) 之后添加的項(xiàng)目流。那么如何將其變成無(wú)限流呢？所以像這樣：public Flux<Item> observe(long id) { return Flux.interval(Duration.ofSeconds(1)). flatMap(counter -> getNew(id)); // <-- how to use last value from getNew flux as the new id }我能夠做到的唯一方法是使用某種類型的狀態(tài)變量： public Flux<Long> observe(long id) { final AtomicLong last = new AtomicLong(id); return Flux.interval(Duration.ofSeconds(1)). flatMap(l -> getNew(last.get())). doOnNext(last::set); } 有沒(méi)有更慣用的方法來(lái)做到這一點(diǎn)？我試圖為此創(chuàng)建生成器，但我不知道如何實(shí)現(xiàn)它。
1 回答
富國(guó)滬深
TA貢獻(xiàn)1790條經(jīng)驗(yàn) 獲得超9個(gè)贊
如果您可以通過(guò)檢查識(shí)別出最后Item發(fā)出的信號(hào)，則可以使用運(yùn)算符：getNew.expand
? ?public Flux<Item> observe(long id) {
? ? ? ? return getNew(id)
? ? ? ? ? ? ? ? .expand(item -> isLast(item)
? ? ? ? ? ? ? ? ? ? ? ? ? getNew(item.id)
? ? ? ? ? ? ? ? ? ? ? ? : Flux.empty());
? ? }
? ? /**
? ? ?* @return true if the given item is the last item emitted by getNew
? ? ?*/
? ? private boolean isLast(Item item) {
? ? ? ? return // ... snip ...
? ? }
如果您不能通過(guò)檢查來(lái)識(shí)別最后一個(gè)Item，那么您將不得不使用狀態(tài)變量。雖然，我建議使用.deferand.repeat而不是?.interval...
?public Flux<Item> observe(long id) {
? ? ? ? final AtomicLong nextStartId = new AtomicLong(id);
? ? ? ? return Flux.defer(() -> getNew(nextStartId.get()))
? ? ? ? ? ? ? ? .doOnNext(item -> nextStartId.set(item.id))
? ? ? ? ? ? ? ? .repeat();
? ? }
反對(duì)使用的主要原因.interval是：
如果沒(méi)有及時(shí)產(chǎn)生需求，將發(fā)出 onError 信號(hào)
因此，如果 API 花費(fèi)的時(shí)間太長(zhǎng)，或者處理結(jié)果的時(shí)間太長(zhǎng)，流將以錯(cuò)誤結(jié)束。對(duì)于較長(zhǎng)的間隔，這可能不是問(wèn)題，但對(duì)于相對(duì)較短的間隔（例如您的示例中的 1 秒），這可能是一個(gè)問(wèn)題。
如果你想在每次重復(fù)迭代之前延遲，那么你可以使用.repeatWhen, 帶有 reactor-extra 的Repeat固定退避。這將為您提供“固定延遲”語(yǔ)義，而不是“固定間隔”。
添加回答
舉報(bào)
0/150
提交
取消