CompletableFuture与Spring的Sleuth结合工具类

Stella981
• 阅读 470

系列目录：

  1. Spring WebFlux运用中的思考与对比
  2. CompletableFuture与Spring的Sleuth结合工具类
  3. CommpetableFuture使用anyOf过程中的一些优化思考
  4. 结合CompletableFuture与Spring的Sleuth结合工具类与allOf以及anyOf

本文基于JDK 11 and JDK 12

按照上一篇内容的分析，我们想在异步代码保留原有的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)) {
                return fn.apply(throwable);
            }
        }, executor), this.tracer);
    }

这样使用和CompletableFuture完全一样，并且：

Mono.fromFuture(CompletableFutureWithSpan)

也是没没有问题的

executorcompletablefuturetry前端前端
点赞
收藏
评论区
推荐文章
blmius blmius
•
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时，报错：原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等，这样可以更容易地在不同的环境中使用MySQL。全局s
3550
0
0
Wesley13 Wesley13
•
2年前
java将前端的json数组字符串转换为列表
记录下在前端通过ajax提交了一个json数组的字符串，在后端如何转换为列表。前端数据转化与请求varcontracts{id:'1',name:'yanggb合同1'},{id:'2',name:'yanggb合同2'},{id:'3',name:'yang
1017
0
0
Jacquelyn38 Jacquelyn38
•
2年前
2020年前端实用代码段，为你的工作保驾护航
有空的时候，自己总结了几个代码段，在开发中也经常使用，谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
2026
0
0
helloworld_34035044 helloworld_34035044
•
1年前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时，如何在新增行里自动增加id呢？能新增整数排序id吗？目前可以在新增行里自动增加id，但只能用uuid函数增加UUID编码，不能新增整数排序id。uuid函数说明：获取一个UUID，可以在填报表中用来创建数据ID语法：uuid()或uuid(sep)参数说明：sep布尔值，生成的uuid中是否包含分隔符''，缺省为
806
0
0
Stella981 Stella981
•
2年前
Spring WebFlux运用中的思考与对比
系列目录：1.SpringWebFlux运用中的思考与对比(https://www.oschina.net/action/GoToLink?urlhttps%3A%2F%2Fblog.csdn.net%2Fzhxdick%2Farticle%2Fdetails%2F103035149)2.CompletableFuture与Sprin
995
0
0
Stella981 Stella981
•
2年前
CommpetableFuture使用anyOf过程中的一些优化思考
系列目录：1.SpringWebFlux运用中的思考与对比(https://www.oschina.net/action/GoToLink?urlhttps%3A%2F%2Fblog.csdn.net%2Fzhxdick%2Farticle%2Fdetails%2F103035149)2.CompletableFuture与Sprin
707
0
0
Wesley13 Wesley13
•
2年前
00：Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork)，斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟，已经成为Web应用程序的首选开发语言。Java是简单易学，完全面向对象，安全可靠，与平台无关的编程语言。
937
0
0
Stella981 Stella981
•
2年前
Django中Admin中的一些参数配置
设置在列表中显示的字段，id为django模型默认的主键list_display('id','name','sex','profession','email','qq','phone','status','create_time')设置在列表可编辑字段list_editable
879
0
0
Wesley13 Wesley13
•
2年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
1148
0
0
Python进阶者 Python进阶者
•
4个月前
Excel中这日期老是出来00：00：00，怎么用Pandas把这个去除
大家好，我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下：这日期老是出来00：00：00，怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下：pd.toexcel之前把这
157
0
0
Stella981
Stella981
Lv1
接天莲叶无穷碧，映日荷花别样红。
文章
1.7w
粉丝
37
获赞
28
热门文章
OpenVPN下载、安装、配置及使用详解
GitHub神器，一个可以白嫖全网无损音乐的神器
OpenWrt 路由器过滤广告的N种方法
Python—执行系统命令的四种方法（os.system、os.popen、commands、subprocess）
SS端加密以及obfs混淆