【问题标题】:Avoid GC pressure with Reactor Publisher使用 Reactor Publisher 避免 GC 压力
【发布时间】:2020-06-05 20:08:51
【问题描述】:

我正在寻找一种方法,通过在每次调用时实例化整个管道来正确使用 Project Reactor 中的发布者,而不会产生无用的 GC 压力。

在我们的代码中,响应服务间 HTTP 请求的典型句柄函数如下所示:

final List<Function<ChangeEvent, Mono<Void>>> triggerOtherMicroservices;

@PostMapping("/handle")
public Mono<Void> handle(@RequestBody ChangeEvent changeEvent) {
    return Mono
            .defer(() -> someService.callToAnotherMicroServiceToFetchData(changeEvent))
            .subscribeOn(Schedulers.parallel())
            .map(this::mapping)
            .flatMap(data -> databaseService.save(data))
            .thenMany(Flux.fromIterable(triggerOtherMicroservices).flatMap(t -> t.apply(changeEvent)))
            .then();
}

如果我理解正确,这意味着在每次调用 handle 时,都需要实例化整个管道(通常具有巨大的堆栈跟踪)(并因此稍后收集)。

我的问题是:我能否以某种方式“准备”整个流程一次并在以后重用它?

我在想Mono.create( ... ) .... 之类的东西。或者,我完全错了,这里不需要考虑优化?

编辑:

进一步思考我可以做的:

final List<Function<ChangeEvent, Mono<Void>>> triggerOtherMicroservices;

final Mono<Void> mono = Mono
        .defer(() -> Mono
                .subscriberContext()
                .map(context -> context.get("event"))
                .flatMap(event -> someService.callToAnotherMicroServiceToFetchData(event))
        )
        .subscribeOn(Schedulers.parallel())
        .flatMap(data -> databaseService.save(data))
        .thenMany(Mono
                .subscriberContext()
                .map(context -> context.get("event"))
                .flatMap(event -> Flux
                        .fromIterable(triggerOtherMicroservices)
                        .flatMap(t -> t.apply(event)))
        )
        .then(); 

public Mono<Void> handle(@Validated ChangeEvent changeEvent) throws NoSuchElementException {
    return mono.subscriberContext(context -> context.put("event", changeEvent));
}

无论如何,我怀疑这就是subscriberContext 的用途。

【问题讨论】:

  • 这是一个理论问题还是您遇到了问题?
  • 我能想到的一个优化是用引用的方法替换 lambda。但不知道这对内存占用有多大影响。
  • @BenjaminEckardt 好吧​​,我不会称之为理论。这是一个与高吞吐量、低延迟优化有关的问题。没有功能问题,只是我们生产中的 cpu/内存利用率。但我的目标不是一般的 java 优化,比如替换 lambdas(据我所知,在较新的 JVM 中应该不会真正产生影响)。问题与反应器和 GC 压力有关。

标签: java reactive-programming spring-webflux project-reactor


【解决方案1】:

注意:有很多 JVM 实现,这个答案并没有声称已经测试了所有这些,也不是所有可能情况的一般性陈述。

根据https://www.bettercodebytes.com/the-cost-of-object-creation-in-java-including-garbage-collection/,当对象只存在于方法中时,可能没有对象创建的开销。这是因为 JIT 实际上并不实例化对象,而是直接执行包含的方法。 因此,以后也不需要垃圾收集。

结合问题的测试可以这样实现:

控制器:

final List<Function<Event, Mono<Void>>> triggerOtherMicroservices = Arrays.asList(
        event -> Mono.empty(),
        event -> Mono.empty(),
        event -> Mono.empty()
);

final Mono<Void> mono = Mono
        .defer(() -> Mono
                .subscriberContext()
                .<Event>map(context -> context.get("event"))
                .flatMap(this::fetch)
        )
        .subscribeOn(Schedulers.parallel())
        .flatMap(this::duplicate)
        .flatMap(this::duplicate)
        .flatMap(this::duplicate)
        .flatMap(this::duplicate)
        .thenMany(Mono
                .subscriberContext()
                .<Event>map(context -> context.get("event"))
                .flatMapMany(event -> Flux
                        .fromIterable(triggerOtherMicroservices)
                        .flatMap(t -> t.apply(event))
                )
        )
        .then();

@PostMapping("/event-prepared")
public Mono<Void> handle(@RequestBody @Validated Event event) throws NoSuchElementException {
    return mono.subscriberContext(context -> context.put("event", event));
}

@PostMapping("/event-on-the-fly")
public Mono<Void> handleOld(@RequestBody @Validated Event event) throws NoSuchElementException {
    return Mono
            .defer(() -> fetch(event))
            .subscribeOn(Schedulers.parallel())
            .flatMap(this::duplicate)
            .flatMap(this::duplicate)
            .flatMap(this::duplicate)
            .flatMap(this::duplicate)
            .thenMany(Flux.fromIterable(triggerOtherMicroservices).flatMap(t -> t.apply(event)))
            .then();
}


private Mono<Data> fetch(Event event) {
    return Mono.just(new Data(event.timestamp));
}

private Mono<Data> duplicate(Data data) {
    return Mono.just(new Data(data.a * 2));
}

数据:

long a;

public Data(long a) {
    this.a = a;
}

@Override
public String toString() {
    return "Data{" +
            "a=" + a +
            '}';
}

事件:

 @JsonSerialize(using = EventSerializer.class)
 public class Event {
     UUID source;
     long timestamp;

     @JsonCreator
     public Event(@JsonProperty("source") UUID source, @JsonProperty("timestamp") long timestamp) {
         this.source = source;
         this.timestamp = timestamp;
     }

     @Override
     public String toString() {
         return "Event{" +
                 "source=" + source +
                 ", timestamp=" + timestamp +
                 '}';
     }
 }

事件序列化器:

 public class EventSerializer extends StdSerializer<Event> {

     public EventSerializer() {
         this(null);
     }

     public EventSerializer(Class<Event> t) {
         super(t);
     }

     @Override
     public void serialize(Event value, JsonGenerator jsonGenerator, SerializerProvider provider) throws IOException {
         jsonGenerator.writeStartObject();
         jsonGenerator.writeStringField("source", value.source.toString());
         jsonGenerator.writeNumberField("timestamp", value.timestamp);
         jsonGenerator.writeEndObject();
     }
 }

最后是测试本身:

 @SpringBootTest
 @AutoConfigureWebTestClient
 class MonoAssemblyTimeTest {

     @Autowired
     private WebTestClient webTestClient;

     final int number_of_requests = 500000;

     @Test
     void measureExecutionTime() throws IOException {
         measureExecutionTime("on-the-fly");
         measureExecutionTime("prepared");
     }

     private void measureExecutionTime(String testCase) throws IOException {
         warmUp("/event-" + testCase);

         final GCStatisticsDifferential gcStatistics = new GCStatisticsDifferential();
         long[] duration = benchmark("/event-" + testCase);

         StringBuilder output = new StringBuilder();
         int plotPointsInterval = (int) Math.ceil((float) number_of_requests / 1000);

         for (int i = 0; i < number_of_requests; i++) {
             if (i % plotPointsInterval == 0) {
                 output.append(String.format("%d , %d %n", i, duration[i]));
             }
         }

         Files.writeString(Paths.get(testCase + ".txt"), output.toString());

         long totalDuration = LongStream.of(duration).sum();
         System.out.println(testCase + " duration: " + totalDuration / 1000000 + " ms.");
         System.out.println(testCase + " average: " + totalDuration / number_of_requests + " ns.");
         System.out.println(testCase + ": " + gcStatistics.get());
     }

     private void warmUp(String path) {
         UUID source = UUID.randomUUID();
         IntStream.range(0, number_of_requests).forEach(i -> call(new Event(source, i), path));
         System.out.println("done with warm-up for path: " + path);
     }

     private long[] benchmark(String path) {
         long[] duration = new long[number_of_requests];

         UUID source = UUID.randomUUID();
         IntStream.range(0, number_of_requests).forEach(i -> {
             long start = System.nanoTime();
             call(new Event(source, i), path).returnResult().getResponseBody();
             duration[i] = System.nanoTime() - start;
         });
         System.out.println("done with benchmark for path: " + path);
         return duration;
     }

     private WebTestClient.BodySpec<Void, ?> call(Event event, String path) {
         return webTestClient
                 .post()
                 .uri(path)
                 .contentType(MediaType.APPLICATION_JSON)
                 .bodyValue(event)
                 .exchange()
                 .expectBody(Void.class);
     }

     private static class GCStatisticsDifferential extends GCStatistics {

         GCStatistics old = new GCStatistics(0, 0);

         public GCStatisticsDifferential() {
             super(0, 0);
             calculateIncrementalGCStats();
         }

         public GCStatistics get() {
             calculateIncrementalGCStats();
             return this;
         }

         private void calculateIncrementalGCStats() {
             long timeNew = 0;
             long countNew = 0;

             for (GarbageCollectorMXBean gc : ManagementFactory.getGarbageCollectorMXBeans()) {

                 long count = gc.getCollectionCount();

                 if (count >= 0) {
                     countNew += count;
                 }

                 long time = gc.getCollectionTime();

                 if (time >= 0) {
                     timeNew += time;
                 }
             }

             time = timeNew - old.time;
             count = countNew - old.count;

             old = new GCStatistics(timeNew, countNew);
         }

     }

     private static class GCStatistics {
         long count, time;

         public GCStatistics(long count, long time) {
             this.count = count;
             this.time = time;
         }

         @Override
         public String toString() {
             return "GCStatistics{" +
                     "count=" + count +
                     ", time=" + time +
                     '}';
         }
     }

 }

结果并不总是相同,但“即时”方法始终优于“准备”方法。另外,“on-the-fly”方法的垃圾回收量要少得多。

典型的结果如下:

完成路径的热身:/event-on-the-fly

完成路径基准测试:/event-on-the-fly

动态持续时间:42679 毫秒。

动态平均值:85358 ns。

即时:GCStatistics{count=29, time=128}

完成了路径的预热:/event-prepared

完成路径基准测试:/event-prepared

准备持续时间:44678 毫秒。

准备好的平均值:89357 ns。

准备:GCStatistics{count=86, time=67}

此结果是在 MacBook Pro(16 英寸,2019 年)、2.4 GHz 8 核 Intel Core i9、64 GB 2667 MHz DDR4 上完成的。

注意:仍然非常欢迎评论、更好的答案或...。

【讨论】:

    【解决方案2】:

    首先,进行一些测量以确定 GC 压力是否真的很高并且值得打扰。

    然后,使用一些面向对象的库,它允许您显式地创建管道对象,并将其重用于多个请求。例如,看看 Vert.x(我从不使用它)。我的库 Df4j 允许创建和重用任何拓扑的计算图,不仅是线性管道,而且它不包含执行 HTTP 请求的模块。但是,Df4j 实现了响应式流协议,因此可以连接到任何兼容的网络库。

    【讨论】:

    • 好的,谢谢,但目前我不考虑转移到 Flow 包的另一个实现。
    猜你喜欢
    • 1970-01-01
    • 2021-05-13
    • 2014-12-19
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2012-02-21
    相关资源
    最近更新 更多