SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Speeds up and lowers memory consumption

Browse Source
This commit is contained in:
Marcin Grzejszczak
2020-07-22 09:30:44 +02:00
parent 41010c103d
commit ccef96f1b6
2 changed files with 24 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java
View File

@@ -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<Flux<Message<String>>, Fl
@Override
public Flux<Message<String>> apply(Flux<Message<String>> 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<String>, Flux<String>> {
private static final Logger log = LoggerFactory.getLogger(SleuthFunction.class);
static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthFunction");
@Override
public Flux<String> apply(Flux<String> 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<String>, Flux<String>> {
private static final Logger log = LoggerFactory.getLogger(SleuthManualFunction.class);
static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthManualFunction");
@Override
public Flux<String> apply(Flux<String> 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 -> {

18
benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java
View File

@@ -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<ReactiveWebServerInitializedEvent> {
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<R
public Mono<String> 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<R
public Mono<String> 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<R
public Mono<String> 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");