有了EmitterProcessor，我可以使用onNext或sink来获取FluxSink并使用它。EmitterProcessor<Long> emitterProcessor = EmitterProcessor.create(100);
emitterProcessor.onNext(1L);
FluxSink<
浏览 131提问于2020-04-04得票数 1
请参阅如果下游无法跟上， FluxSink.OverflowStrategy BUFFER缓冲所有信号。警告！这会进行无界缓冲，并可能导致OutOfMemoryError。我正在寻找一种将此更改为删除或错误的方法：(请参阅https://projectreactor.io/docs/core/release/api/reactor/core/publisher/FluxSink.OverflowStrategy.html
浏览 3提问于2020-04-30得票数 0
试图在消息可用/准备就绪时将消息推送到上游，并在刷新后关闭连接，而不是使用spring反应通量间隔轮询消息。public Flux<String> statusCheck(@PathVariable("randomId") @NonNull String randomId) {
return Flux.<String>interval(Duration.of
浏览 0提问于2019-01-12得票数 2
回答已采纳
2回答
与反应流一起发出
、、、
String msg = "MSG #" + i++; Thread.sleep(1000);输出：INFO reactor.Flux.Map.1 - onNext(Received: 1)INFO reactor.Flux.Map.1 - onNext(Receiv
浏览 0提问于2020-01-19得票数 2
回答已采纳
一段时间以来，我一直在使用带有内置sink的create an EmitterProcessor，如下所示：FluxSink<String> sink = emitter.sink(FluxSink.OverflowStrategy.LATEST);Flux现在我们看到EmitterProcessor</e
浏览 0提问于2021-03-04得票数 0
如何创建只保留最新n个元素的EmitterProcessor，这样即使没有订阅服务器，它也能工作？现在，我创建了这样一个处理器：外部系统在一天内随机提供温度更新。在该系统的回调中，我这样做： processor.onNext(temp);但是，一旦添加了onNext(...)元素，proce
浏览 0提问于2019-01-28得票数 3
回答已采纳
我尝试使用zip和last运算符来组合三个不同的EmitterProcessor，并处理组合后的结果。除了其中一个处理器不发出项的场景之外，这已经按照下面的定义完美地工作了。我对last(defaultValue)运算符的理解是，当EmitterProcessor完成时，它将发出默认值，而不会发出任何值。这将依次完成zip并发送默认值以及来自其他处理器的任何实际值。EmitterProcessor.zip( emitter2.last(&
浏览 0提问于2020-01-21得票数 0
我们正在测试如何使用EmitterProcessor作为事件总线。> emitterProcessor) { // put this subscription); requestService.accept(data);
FluxSink</
浏览 6提问于2020-04-22得票数 2
回答已采纳
我正在尝试使用Flux.buffer()从数据库批量加载。 用例是，从数据库加载记录可能是“突发的”，我想介绍一个小缓冲区，以便在可能的情况下将负载分组在一起。 我的概念方法是使用某种形式的处理器，发布到它的接收器，让缓冲区，然后订阅和过滤我想要的结果。 我尝试了多种不同的方法(不同类型的处理器，以不同的方式创建过滤后的Mono )。 下面是我到目前为止所得到的--主要是通过跌跌撞撞。 目前，这只返回一个结果，但后续调用会被丢弃(尽管我不确定在哪里)。 class BatchLoadingRepository {
// I've tried all manner of dif
浏览 68提问于2019-03-16得票数 24
回答已采纳
11:49:06.936 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)reactor.Flux.EmitterProcessor.1 - onNext(1)11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:49:06.943 [main] INF
浏览 4提问于2021-03-17得票数 12
回答已采纳
> generateResponses(String request) { final FluxProcessor publish = EmitterProcessor.create().serialize(); final FluxSink<Response2> sink2= publish.sink();
fina
浏览 0提问于2020-02-27得票数 0
每一个，我以前都有代码FluxSink<String> sink = emitter.sink(FluxSink.OverflowStrategy.LATEST);
cancelSink(id, request);例如，当使用rSocket
浏览 0提问于2021-03-03得票数 2
回答已采纳
1回答
天真地，我可以实现这样的东西： private final EmitterProcessor<Object> emitterProcessor;
浏览 0提问于2018-01-12得票数 3
回答已采纳
所以首先我编码如下： .doOnNext(s -> System.out.println("2 " + s)); stream.onNextEmitterProcessor<String> stream = EmitterProcessor.&qu
浏览 3提问于2016-11-28得票数 4
回答已采纳
我有一个spring boot rsocket实现，如果客户端取消或关闭他们的rsocket请求，那么我想取消服务器上的其他订阅注册。在spring boot服务器上的日志中，我可以看到发送或接收到一条取消消息：@MessageMapping("cancel") log.info("Captured cance
浏览 20提问于2020-03-09得票数 1
回答已采纳
.setHeader("spring.cloud.stream.sendto.destination", destinationName).build(); return () -> processor; public EmitterProcessor<Message>> processor(){
return EmitterProce
浏览 3提问于2020-02-02得票数 1
回答已采纳
import reactor.core.publisher.EmitterProcessor
@JvmStaticfun main(args: Array<String>) { .subscribe()
Thread.sleep
浏览 0提问于2017-01-06得票数 1
回答已采纳
让我们考虑下面的代码：public void testFluxCreate() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1);
IntStream.range(1, 1001).forEach(it -> {
浏览 15提问于2017-11-09得票数 1
如果我使用subscribeOn(Schedulers.immediate())或弹性，它可以工作。有什么吗，为什么？ val testPublisher = EmitterProcessor.create<String>().connect()
testPublisher.subscribe { println("subscribe on ${Thread.currentThread().name}&quo
浏览 1提问于2017-01-30得票数 11
回答已采纳
云服务器
对象存储
ICP备案
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号