From ccef96f1b6cbea0c9da26ec28b912cb21035de67 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Wed, 22 Jul 2020 09:30:44 +0200 Subject: [PATCH] Speeds up and lowers memory consumption --- .../SleuthBenchmarkingStreamApplication.java | 27 ++++++++++--------- .../SleuthBenchmarkingSpringWebFluxApp.java | 18 ++++++------- 2 files changed, 24 insertions(+), 21 deletions(-) diff --git a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java index 3d51b3717..f13d6d399 100644 --- a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java +++ b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java @@ -29,6 +29,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; import org.springframework.boot.SpringApplication; @@ -205,15 +206,13 @@ class SimpleReactiveManualFunction implements Function>, Fl @Override public Flux> apply(Flux> input) { - return input - .map(message -> (MessagingSleuthOperator.asFunction(this.tracing, message)) - .andThen(msg -> MessagingSleuthOperator.withSpanInScope(this.tracing, msg, stringMessage -> { - log.info("Hello from simple manual [{}]", stringMessage.getPayload()); - return stringMessage; - })).andThen(msg -> MessagingSleuthOperator.afterMessageHandled(this.tracing, msg, null)) - .andThen(msg -> MessageBuilder.createMessage(msg.getPayload().toUpperCase(), msg.getHeaders())) - .andThen(msg -> MessagingSleuthOperator.handleOutputMessage(this.tracing, msg)) - .apply(message)); + return input.map(message -> (MessagingSleuthOperator.asFunction(this.tracing, message)) + .andThen(msg -> MessagingSleuthOperator.withSpanInScope(this.tracing, msg, stringMessage -> { + log.info("Hello from simple manual [{}]", stringMessage.getPayload()); + return stringMessage; + })).andThen(msg -> MessagingSleuthOperator.afterMessageHandled(this.tracing, msg, null)) + .andThen(msg -> MessageBuilder.createMessage(msg.getPayload().toUpperCase(), msg.getHeaders())) + .andThen(msg -> MessagingSleuthOperator.handleOutputMessage(this.tracing, msg)).apply(message)); } } @@ -248,10 +247,12 @@ class SleuthFunction implements Function, Flux> { private static final Logger log = LoggerFactory.getLogger(SleuthFunction.class); + static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthFunction"); + @Override public Flux apply(Flux input) { return input.doOnEach(signal -> log.info("Got a message")) - .flatMap(s -> Mono.delay(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(aLong -> { + .flatMap(s -> Mono.delay(Duration.ofSeconds(1), SCHEDULER).map(aLong -> { log.info("Logging [{}] from flat map", s); return s.toUpperCase(); })); @@ -263,10 +264,12 @@ class SleuthManualFunction implements Function, Flux> { private static final Logger log = LoggerFactory.getLogger(SleuthManualFunction.class); + static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthManualFunction"); + @Override public Flux apply(Flux input) { - return input.doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a message"))).flatMap(s -> Mono - .subscriberContext().delayElement(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(ctx -> { + return input.doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a message"))) + .flatMap(s -> Mono.subscriberContext().delayElement(Duration.ofSeconds(1), SCHEDULER).map(ctx -> { WebFluxSleuthOperators.withSpanInScope(ctx, () -> log.info("Logging [{}] from flat map", s)); return s.toUpperCase(); })).doOnEach(signal -> { diff --git a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java index 936647e73..c93f5dab7 100644 --- a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java +++ b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java @@ -27,6 +27,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; import org.springframework.beans.factory.annotation.Value; @@ -52,6 +53,8 @@ import org.springframework.web.bind.annotation.RestController; @RestController public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener { + static final Scheduler FOO_SCHEDULER = Schedulers.newParallel("foo"); + private static final Logger log = LoggerFactory.getLogger(SleuthBenchmarkingSpringWebFluxApp.class); /** @@ -106,7 +109,7 @@ public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener complexNoSleuth() { return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList()) .doOnEach(signal -> log.info("Got a request")) - .flatMap(s -> Mono.delay(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(aLong -> { + .flatMap(s -> Mono.delay(Duration.ofSeconds(1), FOO_SCHEDULER).map(aLong -> { log.info("Logging [{}] from flat map", s); return ""; })); @@ -116,7 +119,7 @@ public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener complex() { return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList()) .doOnEach(signal -> log.info("Got a request")) - .flatMap(s -> Mono.delay(Duration.ofMillis(1), Schedulers.newParallel("foo")).map(aLong -> { + .flatMap(s -> Mono.delay(Duration.ofSeconds(1), FOO_SCHEDULER).map(aLong -> { log.info("Logging [{}] from flat map", s); return ""; })).doOnEach(signal -> { @@ -132,13 +135,10 @@ public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener complexManual() { return Flux.range(1, 10).map(String::valueOf).collect(Collectors.toList()) .doOnEach(WebFluxSleuthOperators.withSpanInScope(() -> log.info("Got a request"))) - .flatMap(s -> Mono.subscriberContext().delayElement(Duration.ofMillis(1), Schedulers.newParallel("foo")) - .map(ctx -> { - WebFluxSleuthOperators.withSpanInScope(ctx, - () -> log.info("Logging [{}] from flat map", s)); - return ""; - })) - .doOnEach(signal -> { + .flatMap(s -> Mono.subscriberContext().delayElement(Duration.ofSeconds(1), FOO_SCHEDULER).map(ctx -> { + WebFluxSleuthOperators.withSpanInScope(ctx, () -> log.info("Logging [{}] from flat map", s)); + return ""; + })).doOnEach(signal -> { WebFluxSleuthOperators.withSpanInScope(signal.getContext(), () -> log.info("Doing assertions")); TraceContext traceContext = signal.getContext().get(TraceContext.class); Assert.notNull(traceContext, "Context must be set by Sleuth instrumentation");