CompletableFuture与Spring的Sleuth结合工具类

简介: CompletableFuture与Spring的Sleuth结合工具类
本文基于JDK 11 and JDK 12



系列目录:

  1. Spring WebFlux运用中的思考与对比
  2. CompletableFuture与Spring的Sleuth结合工具类
  3. CommpetableFuture使用anyOf过程中的一些优化思考
  4. 结合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)

也是没没有问题的

相关文章
|
2月前
|
Java 测试技术 Spring
springboot学习三:Spring Boot 配置文件语法、静态工具类读取配置文件、静态工具类读取配置文件
这篇文章介绍了Spring Boot中配置文件的语法、如何读取配置文件以及如何通过静态工具类读取配置文件。
61 0
springboot学习三:Spring Boot 配置文件语法、静态工具类读取配置文件、静态工具类读取配置文件
|
2月前
|
监控 Java 对象存储
监控与追踪:如何利用Spring Cloud Sleuth和Netflix OSS工具进行微服务调试
监控与追踪:如何利用Spring Cloud Sleuth和Netflix OSS工具进行微服务调试
45 1
|
5月前
|
Java Spring
spring restTemplate 进行http请求的工具类封装
spring restTemplate 进行http请求的工具类封装
224 3
|
7月前
|
消息中间件 SpringCloudAlibaba Java
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(八)Config服务配置+bus消息总线+stream消息驱动+Sleuth链路追踪
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(八)Config服务配置+bus消息总线+stream消息驱动+Sleuth链路追踪
1016 0
|
7月前
|
开发框架 Java 数据库
|
7月前
|
缓存 小程序 Java
别再造反射轮子了,Spring中ReflectionUtils 工具类,应有尽有!
ReflectionUtils是spring针对反射提供的工具类。
|
存储 NoSQL Java
Spring Boot 如何编写一个通用的 Redis 工具类
Spring Boot 如何编写一个通用的 Redis 工具类
427 0
Spring Boot 如何编写一个通用的 Redis 工具类
|
7月前
|
监控 Java Docker
【Spring Cloud Sleuth 分布式链路跟踪】 —— 每天一点小知识
【Spring Cloud Sleuth 分布式链路跟踪】 —— 每天一点小知识
142 0
|
Java Apache Spring
Spring BeanUtils 2、Cglib BeanCopier 3、Apache BeanUtils 4、Apache PropertyUtils 5、Dozer 那么,我们到底应该选择哪种工具类更加合适呢?为什么Java开发手册中提到禁止使用Apache BeanUtils呢
Spring BeanUtils 2、Cglib BeanCopier 3、Apache BeanUtils 4、Apache PropertyUtils 5、Dozer 那么,我们到底应该选择哪种工具类更加合适呢?为什么Java开发手册中提到禁止使用Apache BeanUtils呢
105 0
|
Java Spring
【Spring】org.springframework.util.StringUtils工具类中commaDelimitedListToStringArray的使用
【Spring】org.springframework.util.StringUtils工具类中commaDelimitedListToStringArray的使用
65 0