本文基于JDK 11 and JDK 12
系列目录:
- Spring WebFlux运用中的思考与对比
- CompletableFuture与Spring的Sleuth结合工具类
- CommpetableFuture使用anyOf过程中的一些优化思考
- 结合CompletableFuture与Spring的Sleuth结合工具类与allOf以及anyOf
按照上一篇内容的分析,我们想在异步代码保留原有的spanId和traceId需要在异步调用前,使用:
Span span = tracer.currentSpan(); try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { //执行异步代码 }
每次使用CompletableFuture都要这么写的话,太麻烦了,所以,我们继承,使用代理的设计模式,将所有的Async方法都覆盖:
对于JDK11:
import brave.Span; import brave.Tracer; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; public class CompletableFutureWithSpan<T> extends CompletableFuture<T> { private final CompletableFuture<T> completableFuture; private final Tracer tracer; CompletableFutureWithSpan(CompletableFuture<T> completableFuture, Tracer tracer) { this.completableFuture = completableFuture; this.tracer = tracer; } private static <T> CompletableFutureWithSpan<T> from(CompletableFuture<T> completableFuture, Tracer tracer) { return new CompletableFutureWithSpan(completableFuture, tracer); } public static <U> CompletableFutureWithSpan<U> supplyAsync(Supplier<U> supplier, Tracer tracer) { Span span = tracer.currentSpan(); return from(CompletableFuture.supplyAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return supplier.get(); } }), tracer); } public static <U> CompletableFutureWithSpan<U> supplyAsync(Supplier<U> supplier, Tracer tracer, Executor executor) { Span span = tracer.currentSpan(); return from(CompletableFuture.supplyAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return supplier.get(); } }, executor), tracer); } public static CompletableFuture<Void> runAsync(Runnable runnable, Tracer tracer) { Span span = tracer.currentSpan(); return from(CompletableFuture.runAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { runnable.run(); } }), tracer); } public static CompletableFuture<Void> runAsync(Runnable runnable, Tracer tracer, Executor executor) { Span span = tracer.currentSpan(); return from(CompletableFuture.runAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { runnable.run(); } }, executor), tracer); } @Override public <U> CompletableFutureWithSpan<U> thenApplyAsync(Function<? super T, ? extends U> fn) { Span span = tracer.currentSpan(); return from(completableFuture.thenApplyAsync(t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t); } }), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.thenApplyAsync(t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<Void> thenAcceptAsync(Consumer<? super T> action) { Span span = tracer.currentSpan(); return from(completableFuture.thenAcceptAsync(t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t); } }), this.tracer); } @Override public CompletableFutureWithSpan<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.thenAcceptAsync(t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<Void> thenRunAsync(Runnable action) { Span span = tracer.currentSpan(); return from(completableFuture.thenRunAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.run(); } }), this.tracer); } @Override public CompletableFutureWithSpan<Void> thenRunAsync(Runnable action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.thenRunAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.run(); } }, executor), this.tracer); } @Override public <U, V> CompletableFutureWithSpan<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) { Span span = tracer.currentSpan(); return from(completableFuture.thenCombineAsync(other, (t, u) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t, u); } }), this.tracer); } @Override public <U, V> CompletableFutureWithSpan<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.thenCombineAsync(other, (t, u) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t, u); } }, executor), this.tracer); } @Override public <U> CompletableFutureWithSpan<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) { Span span = tracer.currentSpan(); return from(completableFuture.thenAcceptBothAsync(other, (t, u) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t, u); } }), this.tracer); } @Override public <U> CompletableFutureWithSpan<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.thenAcceptBothAsync(other, (t, u) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t, u); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) { Span span = tracer.currentSpan(); return from(completableFuture.runAfterBothAsync(other, () -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.run(); } }), this.tracer); } @Override public CompletableFutureWithSpan<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.runAfterBothAsync(other, () -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.run(); } }, executor), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) { Span span = tracer.currentSpan(); return from(completableFuture.applyToEitherAsync(other, t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t); } }), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.applyToEitherAsync(other, t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) { Span span = tracer.currentSpan(); return from(completableFuture.acceptEitherAsync(other, t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t); } }), this.tracer); } @Override public CompletableFutureWithSpan<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.acceptEitherAsync(other, t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) { Span span = tracer.currentSpan(); return from(completableFuture.runAfterEitherAsync(other, () -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.run(); } }), this.tracer); } @Override public CompletableFutureWithSpan<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.runAfterEitherAsync(other, () -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.run(); } }, executor), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) { Span span = tracer.currentSpan(); return from(completableFuture.thenComposeAsync(t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t); } }), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.thenComposeAsync(t -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) { Span span = tracer.currentSpan(); return from(completableFuture.whenCompleteAsync((t, throwable) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t, throwable); } }), this.tracer); } @Override public CompletableFutureWithSpan<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.whenCompleteAsync((t, throwable) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { action.accept(t, throwable); } }, executor), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) { Span span = tracer.currentSpan(); return from(completableFuture.handleAsync((t, throwable) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t, throwable); } }), this.tracer); } @Override public <U> CompletableFutureWithSpan<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.handleAsync((t, throwable) -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(t, throwable); } }, executor), this.tracer); } @Override public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier, Executor executor) { Span span = tracer.currentSpan(); return completableFuture.completeAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return supplier.get(); } }, executor); } @Override public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier) { Span span = tracer.currentSpan(); return completableFuture.completeAsync(() -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return supplier.get(); } }); } @Override public boolean isDone() { return completableFuture.isDone(); } @Override public T get() throws InterruptedException, ExecutionException { return completableFuture.get(); } @Override public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return completableFuture.get(timeout, unit); } @Override public T join() { return completableFuture.join(); } @Override public T getNow(T valueIfAbsent) { return completableFuture.getNow(valueIfAbsent); } @Override public boolean complete(T value) { return completableFuture.complete(value); } @Override public boolean completeExceptionally(Throwable ex) { return completableFuture.completeExceptionally(ex); } @Override public <U> CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn) { return completableFuture.thenApply(fn); } @Override public CompletableFuture<Void> thenAccept(Consumer<? super T> action) { return completableFuture.thenAccept(action); } @Override public CompletableFuture<Void> thenRun(Runnable action) { return completableFuture.thenRun(action); } @Override public <U, V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) { return completableFuture.thenCombine(other, fn); } @Override public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) { return completableFuture.thenAcceptBoth(other, action); } @Override public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action) { return completableFuture.runAfterBoth(other, action); } @Override public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) { return completableFuture.applyToEither(other, fn); } @Override public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) { return completableFuture.acceptEither(other, action); } @Override public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action) { return completableFuture.runAfterEither(other, action); } @Override public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) { return completableFuture.thenCompose(fn); } @Override public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) { return completableFuture.whenComplete(action); } @Override public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) { return completableFuture.handle(fn); } @Override public CompletableFuture<T> toCompletableFuture() { return completableFuture.toCompletableFuture(); } @Override public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) { return completableFuture.exceptionally(fn); } @Override public boolean cancel(boolean mayInterruptIfRunning) { return completableFuture.cancel(mayInterruptIfRunning); } @Override public boolean isCancelled() { return completableFuture.isCancelled(); } @Override public boolean isCompletedExceptionally() { return completableFuture.isCompletedExceptionally(); } @Override public void obtrudeValue(T value) { completableFuture.obtrudeValue(value); } @Override public void obtrudeException(Throwable ex) { completableFuture.obtrudeException(ex); } @Override public int getNumberOfDependents() { return completableFuture.getNumberOfDependents(); } @Override public String toString() { return completableFuture.toString(); } @Override public <U> CompletableFuture<U> newIncompleteFuture() { return completableFuture.newIncompleteFuture(); } @Override public Executor defaultExecutor() { return completableFuture.defaultExecutor(); } @Override public CompletableFuture<T> copy() { return completableFuture.copy(); } @Override public CompletionStage<T> minimalCompletionStage() { return completableFuture.minimalCompletionStage(); } @Override public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) { return completableFuture.orTimeout(timeout, unit); } @Override public CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit) { return completableFuture.completeOnTimeout(value, timeout, unit); } }
对于JDK12: 额外覆盖如下几个方法:
@Override public CompletableFuture<T> exceptionallyCompose(Function<Throwable, ? extends CompletionStage<T>> fn) { return completableFuture.exceptionallyCompose(fn); } @Override public CompletableFutureWithSpan<T> exceptionallyAsync(Function<Throwable, ? extends T> fn) { Span span = tracer.currentSpan(); return from(completableFuture.exceptionallyAsync(throwable -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(throwable); } }), this.tracer); } @Override public CompletableFutureWithSpan<T> exceptionallyAsync(Function<Throwable, ? extends T> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.exceptionallyAsync(throwable -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(throwable); } }, executor), this.tracer); } @Override public CompletableFutureWithSpan<T> exceptionallyComposeAsync(Function<Throwable, ? extends CompletionStage<T>> fn) { Span span = tracer.currentSpan(); return from(completableFuture.exceptionallyComposeAsync(throwable -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { return fn.apply(throwable); } }), this.tracer); } @Override public CompletableFutureWithSpan<T> exceptionallyComposeAsync(Function<Throwable, ? extends CompletionStage<T>> fn, Executor executor) { Span span = tracer.currentSpan(); return from(completableFuture.exceptionallyComposeAsync(throwable -> { try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) { CompletableFuture与Spring的Sleuth结合工具类 return fn.apply(throwable); } }, executor), this.tracer); }
这样使用和CompletableFuture完全一样,并且:
Mono.fromFuture(CompletableFutureWithSpan)
也是没没有问题的