Reactor 3 （7）: I/O 处理

转载 作者：知者 更新时间：2024-03-13 00:55:14 26 4

@Test
public void monoRead() {
    Mono.fromCallable(()-> Files.readAllLines(Paths.get("info.txt")))
            .log()
            .subscribe();
}

@Test
    public void FluxRead() throws IOException{
        log.info("--------------------from iterable--------------------------");
        Flux.fromIterable(Files.readAllLines(Paths.get("info.txt")))
                .log()
                .subscribe();

        log.info("--------------------from stream--------------------------");
        Flux.fromStream(Files.lines(Paths.get("info.txt")))
                .log()
                .subscribe();
    }

@Test
    public void baseWrite() throws IOException {

        Flux<String> flux = Flux.fromStream(Files.lines(Paths.get("info.txt")))
                .map(String::toUpperCase)
                .log();

        flux.subscribe(new BaseSubscriber<String>() {
            BufferedWriter bw = Files.newBufferedWriter(Paths.get("newInfo.txt"));

            @SneakyThrows
            @Override
            protected void hookOnNext(String value) {
                bw.write(value + "\n");
            }

            @SneakyThrows
            @Override
            protected void hookOnComplete() {
                bw.flush();
                bw.write("**** do flush **** \n");
                bw.close();
            }
        });
    }

@Test
public void flushWrite() throws IOException {
    Flux<String> flux = Flux.fromStream(Files.lines(Paths.get("info.txt")))
            .map(String::toUpperCase)
            .log();

    flux.subscribe(new BaseSubscriber<String>() {
        BufferedWriter bw = Files.newBufferedWriter(Paths.get("newInfo.txt"));
        private int count = 0;
        @SneakyThrows
        @Override
        protected void hookOnNext(String value) {
            count++;
            bw.write(value + "\n");
            if (count % 2 == 0) {       // 设定行数进行清理缓存写入文件
                bw.write("**** do flush **** \n");
                bw.flush();
            }
        }
        @SneakyThrows
        @Override
        protected void hookOnComplete() {
            bw.close();
        }
    });
}

26 4 0

• reactor - Reactor 会提供远程处理吗？

  我正在尝试找出我们应该在下一个项目中使用 Akka 还是 Reactor。最重要的问题之一是 future 选择的框架是否会提供远程处理。正如我所看到的，Akka 以我们想要的方式提供了这个。 在 G

• project-reactor - Spring Reactor - 等到一个单声道完成然后做下一个单声道

  假设我有一个返回 Mono 的 repository.save(..) 方法。 还可以说我有一个 repository.findByEmail(..) 它返回一个 Mono。 问题: 我希望第一个 M

• project-reactor - Reactor GroupedFlux - 等待完成

  有一个像下面这样的异步发布者，Project Reactor 有没有办法等到整个流完成处理？ 当然，不必添加一个未知持续时间的 sleep ...... @Test public void group

• project-reactor - Project Reactor 测试从未完成

  我创建了一个简单的 Kafka 消费者，它返回 Flux对象(收到的消息)，我正在尝试使用 StepVerifier 对其进行测试. 在我的测试中，我做了这样的事情: Flux flux = cons

• project-reactor - Resilience4j 速率限制器在项目 Reactor 中无法正常工作？

  我目前正在研究resilience4j 库，由于某种原因，以下代码无法按预期工作: @Test public void testRateLimiterProjectReactor() { //

• reactor - Netflix RxJava 与 Spring Reactor

  关闭。这个问题是opinion-based 。目前不接受答案。 想要改进这个问题吗？更新问题，以便 editing this post 可以用事实和引文来回答它。 . 已关闭 7 年前。 Improv

• project-reactor - Project Reactor + flatMap + Multiple onErrorComplete - 未按预期工作

  当多个 onErrorContinue添加到管道 处理从 flatMap 抛出的特定类型的异常 ，异常处理没有按预期工作。 我希望下面的代码应该删除元素 1 到 6，而订阅者应该使用元素 7 到 10

• reactor-netty - 如何在 reactor netty 中将 ChannelInitializer 添加到 TcpServer

  我有一个 NettyServerCustomizer，下一个代码: @Override public HttpServer apply(final HttpServer httpServer)

• reactor-v1.x.x 对比 reactor-v2.x.x

  我们正在评估 reactor 库以便在我们的项目中使用它。我们的项目得到了 spring 上下文的支持。因此，我们需要一个工具来构建具有 spring 支持的事件驱动应用程序。 此外，我们的主要关注领

• project-reactor - Reactor 3.x - 限制 groupBy Flux 的时间

  不管上游的完整性如何，有没有办法强制 groupBy() 生成的 Flux 在一段时间后完成(或类似地，限制“打开”组的最大数量)？我有如下内容: Flux someFastPublisher; so

• webclient - Spring WebClient reactor.netty.internal.shaded.reactor.pool.PoolAcquireTimeoutException

  我正在使用 Spring WebClient 调用休息服务。如下所述的 post 调用代码。 Mono response = client.post()

• python - reactor.connectTCP 可以在 twisted python 中的 reactor.run 之后发生吗？

  我希望在 react 器运行后添加更多协议(protocol)和工厂。我找不到说明这是允许的文档。当我在 reactor.connectTCP 之前运行 reactor.run 时，程序会在工厂中围绕

• Reactor 3 响应式编程中文参考文档

  （译者加）本文档的一些典型的名词如下： Publisher（发布者）、Subscriber（订阅者）、Subscription（订阅 n.）、subscribe（订阅 v.）。 event/signa

• java - Reactor:检测空信号以能够作用于空Mono

  Project Reactor Mono 是否有运算符或一些好的方法来实现 doOnEmpty() 的行为？ 我想对操作结果产生副作用(日志记录)。 这是我现在拥有的: myMono .map(v

• project-reactor - 为助焊剂消费者注册不同的线程

  在以下两个示例中，处理通量流的行为似乎有所不同。 示例 1: public static void main(String[] args) throws InterruptedException {

• project-reactor - 如何使用千分尺获得默认的通量指标？

  是否可以使用 LoggingMeterRegistry 以微米为单位收集项目 react 器的通量指标？ 最佳答案 只需添加 LoggingMeterRegistry 就可以了到全局 Micromet

• project-reactor - 存储先前映射结果的语法

  我想知道在spring web-flux中使用先前映射结果的好方法，例如 Mono.just(request) ... .flatMap(object0 -> createObject1(object

• project-reactor - 从助焊剂转换为单声道

  如何将具有1个元素的助焊剂转换为单声道？ Flux.fromArray(arrayOf(1,2,1,1,1,2)) .distinct()

• kotlin - Reactor-加载分页API的所有页面

  我想知道Reactor和分页的HTTP API。我有一个private fun getPage(pageNumber: Int): Mono。该资源具有“numberOfPages”字段，我想获取所有

• java - Reactor 有没有办法忽略错误信号？

  我有一个包含多个 URL 和端口的数组。对于他们每个人，我需要发送和接收返回的内容: Flux.fromArray(trackersArray) .flatMap(tracker ->