预加载元件用于 Flux.generate(..)

Question

我正在使用Flux.generate((创建一个Flux。生成器(使用者(实际上是从消息队列中读取。问题是这个调用需要相当长的时间(有时甚至 1-2 秒(。这将使助焊剂停止处理。

package com.github.loa.vault.service.listener;
import com.github.loa.document.service.domain.DocumentType;
import com.github.loa.queue.service.QueueManipulator;
import com.github.loa.queue.service.domain.Queue;
import com.github.loa.queue.service.domain.message.DocumentArchivingMessage;
import com.github.loa.vault.service.domain.DocumentArchivingContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.SynchronousSink;
import java.util.function.Consumer;
@Slf4j
@Service
@RequiredArgsConstructor
public class VaultQueueConsumer implements Consumer<SynchronousSink<DocumentArchivingContext>> {
private final QueueManipulator queueManipulator;
@Override
public void accept(final SynchronousSink<DocumentArchivingContext> documentSourceItemSynchronousSink) {
final DocumentArchivingMessage documentArchivingMessage = (DocumentArchivingMessage)
queueManipulator.readMessage(Queue.DOCUMENT_ARCHIVING_QUEUE);
documentSourceItemSynchronousSink.next(
DocumentArchivingContext.builder()
.type(DocumentType.valueOf(documentArchivingMessage.getType()))
.source(documentArchivingMessage.getSource())
.content(documentArchivingMessage.getContent())
.build()
);
}
}

显然，添加parallel无济于事，因为生成器仍然一次调用一个。

Flux.generate(vaultQueueConsumer)
.parallel()
.runOn(Schedulers.parallel()) 
.flatMap(vaultDocumentManager::archiveDocument)
.subscribe();

有人知道如何使发电机并联吗？我不想使用Flux.create()因为那样我会失去背压。

Answer 1

Mono.just(1).repeat()  // create infinite flux, maybe there is a nicer way for that?
.flatMap(this::readFromQueue, 100) // define queue polling concurrency
.flatMap(this::archiveDocument)
.subscribe();

private Mono<String> readFromQueue(Integer ignore)
{
return Mono.fromCallable(() -> {
Thread.sleep(1500); // your actual blocking queue polling here
return "queue_element";
}).subscribeOn(Schedulers.elastic()); // dedicate blocking call to threadpool
}

Answer 2

问题是vaultQueueConsumer包括缓慢的操作。 因此，解决方案是将这种缓慢的操作从generate提取到可以并行化的map。

作为一个想法，您可以生成一个队列名称，其中必须使用消息，并在使通量并行后以map方法执行实际的消息消耗：

String queue = "test";
Flux.<String>generate(synchronousSink -> synchronousSink.next(queue))
.parallel()
.runOn(Schedulers.parallel())
.map(queueManipulator::readMessage)
.doOnNext(log::info)
.subscribe();

假QueueManipulator在返回消息之前休眠 1-2 秒：

public class QueueManipulator {
private final AtomicLong counter = new AtomicLong();
public String readMessage(String queue) {
sleep(); //sleep 1-2 seconds
return queue + " " + counter.incrementAndGet();
}
//...
}

这样，消息使用是并行完成的：

12:49:22.362 [parallel-4] - test 2
12:49:22.362 [parallel-3] - test 4
12:49:22.362 [parallel-2] - test 1
12:49:22.362 [parallel-1] - test 3
12:49:23.369 [parallel-3] - test 6
12:49:23.369 [parallel-1] - test 5
12:49:23.369 [parallel-2] - test 7
12:49:23.369 [parallel-4] - test 8

上面的这个解决方案很简单，购买可能看起来像一个"黑客"。

另一个想法是在flatMap内部调用Flux.generate：

String queue = "test";
int parallelism = 5;
Flux.range(0, parallelism)
.flatMap(i ->
Flux.<String>generate(synchronousSink -> {
synchronousSink.next(queueManipulator.readMessage(queue));
}).subscribeOn(Schedulers.parallel()))
.doOnNext(log::info)
.subscribe();

Answer 3

你试过吗：

Flux.generate(vaultQueueConsumer)
.parallel()
.runOn(Schedulers.parallel()) 
.flatMap(vaultDocumentManager::archiveDocument)
.subscribe();