参考: https://zhuanlan.zhihu.com/p/360172630
https://www.cnblogs.com/luliang888/p/14440118.html

概览

CompletableFuture是JDK8对Future接口的增强

  • 提供了函数式编程写法, 使代码简练, 语义清晰
  • 默认使用forkJoinPool线程池, 无需手工维护线程
  • completionStage接口, 提供了异步线程编排的能力, 支持链式编程

函数式接口

有且仅有一个抽象方法的接口为函数式接口, 可以使用@FunctionalInterface标识, 函数式接口能适用于函数式编程场景(Lambda).

常用的函数式接口

接口 说明
Runnable 无参无返回值
Fuction 接收T类型参数, 返回R类型结果
Consumer 接收T类型参数, 无返回值
Predicate 接收T类型参数, 返回boolean类型结果
Supplier 无参, 返回T类型的结果

ComletableFuture使用

接口概览

CompletableFuture最佳实践 - 图1

创建CompletableFuture对象

  1. //使用默认线程池
  2. static CompletableFuture<Void> runAsync(Runnable runnable)
  3. static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
  4. //可以指定线程池  
  5. static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
  6. static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
  • 带Async后缀的方法为异步执行, 下同
  • Runnable接口无返回值, 而Supplier接口的get()是有返回值的
  • 默认会使用 ForkJoinPool 线程池，这个线程池默认创建的线程数是 CPU 的核数
  • 如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议根据不同的业务类型创建不同的线程池，以避免互相干扰

使用CompletionStage编排异步任务

串行关系

  1. CompletionStage<R> thenApply(fn);
  2. CompletionStage<R> thenApplyAsync(fn);
  3. CompletionStage<Void> thenAccept(consumer);
  4. CompletionStage<Void> thenAcceptAsync(consumer);
  5. CompletionStage<Void> thenRun(action);
  6. CompletionStage<Void> thenRunAsync(action);
  7. CompletionStage<R> thenCompose(fn);
  8. CompletionStage<R> thenComposeAsync(fn);
  • 使用上的区别为方法参数的函数式接口, 是否接收参数, 以及是否有返回值

  • thenApply与thenCompose区别

    • thenApply转换的是泛型中的类型, 相当于将CompletableFuture 转换生成新的CompletableFuture

    • thenCompose用来连接两个CompletableFuture，生成一个新的CompletableFuture

      1. public void serial(){
      2.    CompletableFuture<String> f0 = CompletableFuture.supplyAsync(() -> "hello!");
      3.    CompletableFuture<String> f1 = f0.thenApply(str -> str + "world!");
      4.    System.out.println(f1.join()); // hello!world!
      6.    CompletableFuture<String> f2 = f0
      7.            .thenCompose(str -> CompletableFuture.supplyAsync(() -> str + "world!"));
      8.    System.out.println(f2.join()); // hello!world!
      9. }

汇聚关系

AND
  1. CompletionStage<R> thenCombine(other, fn);
  2. CompletionStage<R> thenCombineAsync(other, fn);
  3. CompletionStage<Void> thenAcceptBoth(other, consumer);
  4. CompletionStage<Void> thenAcceptBothAsync(other, consumer);
  5. CompletionStage<Void> runAfterBoth(other, action);
  6. CompletionStage<Void> runAfterBothAsync(other, action);
  • 主要区别为方法参数函数式接口的不同
    1. public void and(){
    2.       CompletableFuture<String> f0 = CompletableFuture.supplyAsync(() -> "es recall!");
    3.       CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "sm recall!");
    4.       String res = f0.thenCombine(f1, (str1, str2) -> str1 + " & " + str2).join();
    5.       System.out.println(res); // es recall! & sm recall!
    6.   }

OR
  1. CompletionStage applyToEither(other, fn);
  2. CompletionStage applyToEitherAsync(other, fn);
  3. CompletionStage acceptEither(other, consumer);
  4. CompletionStage acceptEitherAsync(other, consumer);
  5. CompletionStage runAfterEither(other, action);
  6. CompletionStage runAfterEitherAsync(other, action);
  • 主要区别为方法参数函数式接口的不同

  • OR在语义上理解为”最快返回”, 拿最快返回的结果作为下一次任务的输入

    1. public void or() {
    2.       CompletableFuture<String> f0 = CompletableFuture.supplyAsync(() -> {
    3.           try {
    4.               Thread.sleep(random.nextInt(3));
    5.           } catch (InterruptedException e) {
    6.               e.printStackTrace();
    7.           }
    8.           return "es recall!";
    9.       });
    11.       CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
    12.           try {
    13.               Thread.sleep(random.nextInt(3));
    14.           } catch (InterruptedException e) {
    15.               e.printStackTrace();
    16.           }
    17.           return "sm recall!";
    18.       });
    19.       String res = f0.applyToEitherAsync(f1, str -> "fastest: " + str).join();
    20.       System.out.println(res); // fastest: es recall! 或者 fastest: sm recall!
    21.   }

异常处理

  1. CompletionStage exceptionally(fn);
  • 相当于try{}catch{}中的catch{}, 支持链式编程
    1.   public void exception(){
    2.       CompletableFuture<Object> f0 = CompletableFuture.supplyAsync(() -> {
    3.           throw new RuntimeException("hello!exception!");
    4.       }).exceptionally(e -> {
    5.           e.printStackTrace();
    6.           return "exception happened";
    7.       });
    8.       System.out.println(f0.join()); //exception happened
    9.   }

任务结束

  1. CompletionStage<R> whenComplete(consumer);
  2. CompletionStage<R> whenCompleteAsync(consumer);
  3. CompletionStage<R> handle(fn);
  4. CompletionStage<R> handleAsync(fn);
  • 相当于finally{}, 无论异常与否, 都会执行 consumer/fn 回调函数
    1.   public void complete() {
    2.       CompletableFuture<Object> f0 = CompletableFuture.supplyAsync(() -> {
    3.           throw new RuntimeException("hello!exception!");
    4.       }).whenComplete((str, e) -> {
    5.           if (e != null) {
    6.               e.printStackTrace();
    7.           } else {
    8.               System.out.println(str);
    9.           }
    10.       });
    11.   }

超时

  1. /**
  2.  * java8中CompletableFuture异步处理超时的方法
  3.  * <p>
  4.  * Java 8 的 CompletableFuture 并没有 timeout 机制，虽然可以在 get 的时候指定 timeout，是一个同步堵塞的操作。怎样让 timeout 也是异步的呢？Java 8 内有内建的机
  5.  * 制支持，一般的实现方案是启动一个 ScheduledThreadpoolExecutor 线程在 timeout 时间后直接调用 CompletableFuture.completeExceptionally(new TimeoutException())，
  6.  * 然后用acceptEither() 或者 applyToEither 看是先计算完成还是先超时：
  7.  * <p>
  8.  * 在 java 9 引入了 orTimeout 和 completeOnTimeOut 两个方法支持 异步 timeout 机制：
  9.  * <p>
  10.  * public CompletableFuture orTimeout(long timeout, TimeUnit unit) : completes the CompletableFuture with a TimeoutException after the specified timeout has elapsed.
  11.  * public CompletableFuture completeOnTimeout(T value, long timeout, TimeUnit unit) : provides a default value in the case that the CompletableFuture pipeline times out.
  12.  * 内部实现上跟我们上面的实现方案是一模一样的，只是现在不需要自己实现了。
  13.  * <p>
  14.  * 实际上hystrix等熔断的框架，其实现线程Timeout之后就关闭线程，也是基于同样的道理，所以我们可以看到hystrix中会有一个Timer Thread
  15.  * 参考: https://www.cnblogs.com/luliang888/p/14440118.html
  16.  *
  17.  * @author xinzhang
  18.  * @version 2022/5/10
  19.  */
  20. public class CompletableFutureTimeout<T> {
  22.     /**
  23.      * Singleton delay scheduler, used only for starting and * cancelling tasks.
  24.      */
  25.     static final class Delayer {
  27.         static ScheduledFuture<?> delay(Runnable command, long delay,
  28.                 TimeUnit unit) {
  29.             return delayer.schedule(command, delay, unit);
  30.         }
  32.         static final class DaemonThreadFactory implements ThreadFactory {
  34.             @Override
  35.             public Thread newThread(Runnable r) {
  36.                 Thread t = new Thread(r);
  37.                 t.setDaemon(true);
  38.                 t.setName("CompletableFutureDelayScheduler");
  39.                 return t;
  40.             }
  41.         }
  43.         static final ScheduledThreadPoolExecutor delayer;
  45.         static {
  46.             (delayer = new ScheduledThreadPoolExecutor(
  47.                     1, new CompletableFutureTimeout.Delayer.DaemonThreadFactory())).
  48.                     setRemoveOnCancelPolicy(true);
  49.         }
  50.     }
  52.     /**
  53.      * timeout时间后抛出TimeoutException
  54.      */
  55.     private static <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) {
  56.         CompletableFuture<T> result = new CompletableFuture<T>();
  57.         CompletableFutureTimeout.Delayer.delayer
  58.                 .schedule(() -> result.completeExceptionally(new TimeoutException()), timeout, unit);
  59.         return result;
  60.     }
  62.     /**
  63.      * future执行超时返回默认值
  64.      */
  65.     public static <T> CompletableFuture<T> completeOnTimeout(T t, CompletableFuture<T> future, long timeout,
  66.             TimeUnit unit) {
  67.         final CompletableFuture<T> timeoutFuture = timeoutAfter(timeout, unit);
  68.         return future.applyToEither(timeoutFuture, Function.identity()).exceptionally((throwable) -> t);
  69.     }
  71.     /**
  72.      * future执行超时抛出异常
  73.      */
  74.     public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) {
  75.         final CompletableFuture<T> timeoutFuture = timeoutAfter(timeout, unit);
  76.         return future.applyToEither(timeoutFuture, Function.identity());
  77.     }
  78. }

使用示例

  1.     public void completeOnTimeout() {
  2.         CompletableFuture<String> f0 = CompletableFuture.supplyAsync(() -> {
  3.             try {
  4.                 Thread.sleep(2000);
  5.             } catch (InterruptedException e) {
  6.                 e.printStackTrace();
  7.             }
  8.             return "executing 2000 ms";
  9.         });
  11.         CompletableFuture<String> within = CompletableFutureTimeout
  12.                 .completeOnTimeout("timeout default value", f0, 1, TimeUnit.SECONDS);
  13.         System.out.println(within.join()); //timeout default value
  14.     }

其他

  1. // 显式返回执行异常
  2. boolean completeExceptionally(Throwable ex)
  3. // 全部完成
  4. CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
  5. // 其中一个完成即返回
  6. CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

使用CompletionService批量执行异步任务

CompletionService可以用于大量独立同构任务的异步批量执行, 可以submit提交任务, 通过take/poll获取任务Future结果
接口概览:

  1. Future<V> submit(Callable<V> task);
  2. Future<V> submit(Runnable task, V result);
  3. Future<V> take() throws InterruptedException;
  4. Future<V> poll();
  5. Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
  • 提交的任务互相独立执行, 谁先完成先返回
  • take()、poll() 都是从阻塞队列中获取并移除一个元素; 它们的区别在于如果阻塞队列是空的，那么调用 take() 方法的线程会被阻塞，而 poll() 方法会返回 null 值

CompletionService接口的实现是ExecutorCompletionService, 其实现原理是其内部维护了一个阻塞队列, 该阻塞队列用来保存任务执行结果的Future对象
ExecutorCompletionService构造方法

  1. // 如果不指定 completionQueue，那么默认会使用无界的 LinkedBlockingQueue
  2. ExecutorCompletionService(Executor executor)
  3. ExecutorCompletionService(Executor executor,BlockingQueue<Future<V>> completionQueue)

示例

  1.  private CompletionService<String> completionService = new ExecutorCompletionService<>(
  2.             Executors.newFixedThreadPool(3));
  4.     public void batchExecute() throws ExecutionException, InterruptedException {
  5.         completionService.submit(() -> "hello!");
  6.         completionService.submit(() -> "world!");
  7.         completionService.submit(() -> "nice!");
  9.         for (int i = 0; i < 3; i++) {
  10.             System.out.println(completionService.take().get());
  11.         }
  12.     }

高级主题

线程池配置

  1. /**
  2.  * ThreadPoolConfig
  3.  *
  4.  * @author xinzhang
  5.  * @version 2022/5/10
  6.  */
  7. @Configuration
  8. public class ThreadPoolConfig {
  10.     @Bean
  11.     public ThreadPoolExecutorFactoryBean bizExecutor() {
  12.         ThreadPoolExecutorFactoryBean factoryBean = new ThreadPoolExecutorFactoryBean();
  13.         // 核心线程数,一直存活
  14.         factoryBean.setCorePoolSize(5);
  15.         // 当线程数大于或等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。
  16.         // 如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池会拒绝处理任务而抛出异常。
  17.         factoryBean.setMaxPoolSize(10);
  18.         // 任务队列容量
  19.         factoryBean.setQueueCapacity(20);
  20.         // 当线程空闲时间达到setKeepAliveSeconds，该线程会退出，直到线程数量等于corePoolSize。
  21.         factoryBean.setKeepAliveSeconds(60);
  22.         factoryBean.setThreadNamePrefix("biz-task");
  23.         //(1) 默认的ThreadPoolExecutor.AbortPolicy   处理程序遭到拒绝将抛出运行时RejectedExecutionException;
  24.         //(2) ThreadPoolExecutor.CallerRunsPolicy 线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度
  25.         //(3) ThreadPoolExecutor.DiscardPolicy  不能执行的任务将被删除;
  26.         //(4) ThreadPoolExecutor.DiscardOldestPolicy  如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。
  27.         factoryBean.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  28.         return factoryBean;
  29.     }
  31. }
  • 线程数量
    • CPU密集型: 多线程本质上是提升多核 CPU 的利用率, 理论上“线程的数量 =CPU 核数”就是最合适的。不过在工程上，线程的数量一般会设置为“CPU 核数 +1”，这样的话，当线程因为偶尔的内存页失效或其他原因导致阻塞时，这个额外的线程可以顶上，从而保证 CPU 的利用率
    • I/O密集型: 最佳线程数 =CPU 核数 * [ 1 +（I/O 耗时 / CPU 耗时）]
  • 不建议使用 Java 并发包中的静态工厂类Executors , 原因是：Executors 提供的很多方法默认使用的都是无界的 LinkedBlockingQueue，高负载情境下，无界队列很容易导致 OOM，而 OOM 会导致所有请求都无法处理，这是致命问题。所以强烈建议使用有界队列。
  • 默认拒绝策略要慎重使用, 默认的拒绝策略会 throw RejectedExecutionException 这是个运行时异常，对于运行时异常编译器并不强制 catch 它，所以开发人员很容易忽略, 在实际工作中，自定义的拒绝策略往往和降级策略配合使用。
  • 注意异常处理, 如果任务在执行的过程中出现运行时异常，会导致执行任务的线程终止；不过，最致命的是任务虽然异常了，但是你却获取不到任何通知，这会让你误以为任务都执行得很正常。最稳妥和简单的方案还是捕获所有异常并按需处理, 如下示例代码 ```java

try { //业务逻辑 } catch (RuntimeException x) { //按需处理 } catch (Throwable x) { //按需处理 } 

  2. 使用示例
  3. ```java
  4.     @Autowired
  5.     private ExecutorService bizExecutor;
  7.     public void runWithExecutor() {
  8.         for (int i = 0; i < 10; i++) {
  9.             CompletableFuture.runAsync(() -> {
  10.                 System.out.println(Thread.currentThread().getName()+"===========>hello!world!");
  11.             }, bizExecutor);
  12.         }
  13.     }

上下文

TransmittableThreadLocal

建议阅读: https://github.com/alibaba/transmittable-thread-local

TransmittableThreadLocal(TTL)：在使用线程池等会池化复用线程的执行组件情况下，提供ThreadLocal值的传递功能，解决异步执行时上下文传递的问题

原理浅析: https://cloud.tencent.com/developer/article/1484420

使用方式同传统的ThreadLocal, 父子线程传值示例:

  1.  public void transferCtx() {
  2.         TransmittableThreadLocal<String> ctx = new TransmittableThreadLocal<>();
  3.         ctx.set("hello!world!");
  5.         new Thread(() -> System.out.println(ctx.get())).run(); // hello!world!
  6.     }

具体在CompletableFuture使用场景上:

方案一: 使用Java Agent修饰JDK线程池实现类

在Java的启动参数加上：-javaagent:path/to/transmittable-thread-local-2.x.y.jar

方案二(推荐): CompletableFuture使用自定义线程池, 并使用TtlExecutors修饰
  1. @Configuration
  2. public class ThreadPoolConfig {
  4.     @Bean
  5.     public ThreadPoolExecutorFactoryBean bizExecutor() {
  6.         ThreadPoolExecutorFactoryBean factoryBean = new ThreadPoolExecutorFactoryBean();
  7.         // 核心线程数,一直存活
  8.         factoryBean.setCorePoolSize(5);
  9.         // 当线程数大于或等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。
  10.         // 如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池会拒绝处理任务而抛出异常。
  11.         factoryBean.setMaxPoolSize(10);
  12.         // 任务队列容量
  13.         factoryBean.setQueueCapacity(20);
  14.         // 当线程空闲时间达到setKeepAliveSeconds，该线程会退出，直到线程数量等于corePoolSize。
  15.         factoryBean.setKeepAliveSeconds(60);
  16.         factoryBean.setThreadNamePrefix("biz-task");
  17.         //(1) 默认的ThreadPoolExecutor.AbortPolicy   处理程序遭到拒绝将抛出运行时RejectedExecutionException;
  18.         //(2) ThreadPoolExecutor.CallerRunsPolicy 线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度
  19.         //(3) ThreadPoolExecutor.DiscardPolicy  不能执行的任务将被删除;
  20.         //(4) ThreadPoolExecutor.DiscardOldestPolicy  如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。
  21.         factoryBean.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  22.         return factoryBean;
  23.     }
  25.     @Bean
  26.     public ExecutorService ttlExecutor(ExecutorService bizExecutor) {
  27.         return TtlExecutors.getTtlExecutorService(bizExecutor);
  28.     }
  30. }

使用示例

  1.  @Autowired
  2.     private ExecutorService ttlExecutor;
  4.     public String decorateExecutor() {
  5.         TransmittableThreadLocal<Long> ctx = new TransmittableThreadLocal<>();
  6.         try {
  7.             Thread.sleep(100);
  8.         } catch (InterruptedException e) {
  9.             e.printStackTrace();
  10.         }
  11.         ctx.set(System.currentTimeMillis());
  12.         return CompletableFuture.supplyAsync(() -> Thread.currentThread().getName() + "=====>" + ctx.get(),
  13.                 ttlExecutor).join();
  14.     }

测试

  1.  @Test
  2.     public void testDecorateExecutor() {
  3.         for (int i = 0; i < 10; i++) {
  4.             System.out.println(demo.decorateExecutor());
  5.         }
  6.     }
  8. ==========================
  9. 15:32:55.649 DEBUG [main] o.springframework.test.context.cache - Spring test ApplicationContext cache statistics: [DefaultContextCache@6f1d799 size = 1, maxSize = 32, parentContextCount = 0, hitCount = 2, missCount = 1]
  10. biz-task1=====>1652167976179
  11. biz-task2=====>1652167976279
  12. biz-task3=====>1652167976379
  13. biz-task4=====>1652167976489
  14. biz-task5=====>1652167976599
  15. biz-task1=====>1652167976699
  16. biz-task2=====>1652167976809
  17. biz-task3=====>1652167976919
  18. biz-task4=====>1652167977019
  19. biz-task5=====>1652167977129

Logback的MDC

方案一: logback-mdc-ttl

建议阅读: https://github.com/ofpay/logback-mdc-ttl

实现上集成使用了Transmittable ThreadLocal(TTL) ：在使用线程池等会缓存线程的组件情况下，提供ThreadLocal值的传递功能，解决异步执行时上下文传递的问题。支持JDK 9/8/7/6。

方案二(推荐): 实现MDCAdaper接口, 使用TransmittableThreadLocal替换默认InheritableThreadLocal实现
  • TtlMDCAdapter仅将2.4.0版本logback的BasicMDCAdapter中的InheritableThreadLocal替换为TransmittableThreadLocal ```java package org.slf4j;

import com.alibaba.ttl.TransmittableThreadLocal; import java.util.HashMap; import java.util.Map; import java.util.Set; import org.slf4j.spi.MDCAdapter;

/**

  • TtlMDCAdapter *
  • @author xinzhang

  • @version 2022/5/10 */ public class TtlMDCAdapter implements MDCAdapter {

    private TransmittableThreadLocal> transmittableThreadLocal = new TransmittableThreadLocal>() {

    1.  @Override
    2.  protected Map<String, String> childValue(Map<String, String> parentValue) {
    3.      return parentValue == null ? null : new HashMap<>(parentValue);
    4.  }

    };

    public TtlMDCAdapter() { }

    @Override public void put(String key, String val) {

    1.  if (key == null) {
    2.      throw new IllegalArgumentException("key cannot be null");
    3.  } else {
    4.      Map<String, String> map = this.transmittableThreadLocal.get();
    5.      if (map == null) {
    6.          map = new HashMap<>();
    7.          this.transmittableThreadLocal.set(map);
    8.      }
    10.      map.put(key, val);
    11.  }

    }

    @Override public String get(String key) {

    1.  Map<String, String> map = this.transmittableThreadLocal.get();
    2.  return map != null && key != null ? map.get(key) : null;

    }

    @Override public void remove(String key) {

    1.  Map<String, String> map = this.transmittableThreadLocal.get();
    2.  if (map != null) {
    3.      map.remove(key);
    4.  }

    }

    @Override public void clear() {

    1.  Map<String, String> map = this.transmittableThreadLocal.get();
    2.  if (map != null) {
    3.      map.clear();
    4.      this.transmittableThreadLocal.remove();
    5.  }

    }

    public Set getKeys() {

    1.  Map<String, String> map = this.transmittableThreadLocal.get();
    2.  return map != null ? map.keySet() : null;

    }

    @Override public Map getCopyOfContextMap() {

    1.  Map<String, String> oldMap = this.transmittableThreadLocal.get();
    2.  return oldMap != null ? new HashMap<>(oldMap) : null;

    }

    @Override public void setContextMap(Map contextMap) {

    1.  this.transmittableThreadLocal.set(new HashMap<>(contextMap));

    } }

  2. - 实例化, **注意包结构固定为org.slf4j.impl**, 通过MDC#bwCompatibleGetMDCAdapterFromBinder()实例化
  3. ```java
  4. package org.slf4j.impl;
  6. import org.slf4j.TtlMDCAdapter;
  7. import org.slf4j.spi.MDCAdapter;
  9. /**
  10.  * StaticMDCBinder
  11.  *
  12.  * @author xinzhang
  13.  * @version 2022/5/10
  14.  */
  15. public class StaticMDCBinder {
  17.     public static final StaticMDCBinder SINGLETON = new StaticMDCBinder();
  19.     private StaticMDCBinder() {
  20.     }
  22.     public MDCAdapter getMDCA() {
  23.         return new TtlMDCAdapter();
  24.     }
  26.     public String getMDCAdapterClassStr() {
  27.         return TtlMDCAdapter.class.getName();
  28.     }
  29. }

使用示例

  1.  @Autowired
  2.     private ExecutorService ttlExecutor;
  4.     public String transferMDC() {
  5.         try {
  6.             Thread.sleep(100);
  7.         } catch (InterruptedException e) {
  8.             e.printStackTrace();
  9.         }
  10.         MDC.put("time", String.valueOf(System.currentTimeMillis()));
  11.         return CompletableFuture.supplyAsync(() -> Thread.currentThread().getName() + "=====>" + MDC.get("time")
  12.         ,ttlExecutor).join();
  13.     }

测试

  1.  @Test
  2.     public void testTransferMDC() {
  3.         for (int i = 0; i < 10; i++) {
  4.             System.out.println(demo.transferMDC());
  5.         }
  6.     }
  7. =========================
  8. biz-task1=====>1652172234438
  9. biz-task2=====>1652172234548
  10. biz-task3=====>1652172234658
  11. biz-task4=====>1652172234758
  12. biz-task5=====>1652172234858
  13. biz-task1=====>1652172234968
  14. biz-task2=====>1652172235078
  15. biz-task3=====>1652172235178
  16. biz-task4=====>1652172235288
  17. biz-task5=====>1652172235398

Hystrix线程池隔离模式

参考: https://zhuanlan.zhihu.com/p/273292662
hystrix默认为线程池隔离模式, 会复用线程, 导致上下文传递出现问题
示例
serviceA—>serviceB

  1. 1.feign集成了hystrix, 在配置中打开开关即可, 将核心线程数限制为3
  3. # hystrix
  4. feign.hystrix.enabled=true
  5. hystrix.threadpool.default.coreSize=3
  6. hystrix.threadpool.default.maxQueueSize=50
  7. # 没达到maxQueueSize, 但达到queueSizeRejectionThreshold值, 请求也会被拒绝, 默认为5
  8. hystrix.threadpool.default.queueSizeRejectionThreshold=20
  10. 2.serviceA调用serviceB
  11. (TtlContext为TransmittableThreadLocal实现的上下文)
  13.  @GetMapping
  14.     public void accessServiceB() {
  15.         long now = System.currentTimeMillis();
  16.         System.out.println(Thread.currentThread().getName() + "设置上下文=======>" + now);
  17.         // 设置上下文
  18.         TtlContext.put(now);
  19.         // 业务操作, 调用serviceB
  20.         serviceBFeignClient.hello();
  21.         // 请求结束, 清除上下文
  22.         TtlContext.clear();
  23.     }
  25. 3.在serviceB的feignClient设置拦截器, 获取上下文
  26. @Component
  27. public class FeignInterceptor implements RequestInterceptor {
  29.     @Override
  30.     public void apply(RequestTemplate requestTemplate) {
  31.         System.out.println(Thread.currentThread().getName() + "获取上下文=======>" + TtlContext.get());
  32.     }
  33. }
  35. 3.用jmeter以10个并发一次访问serviceA, 使hystrix线程复用, 日志如下:
  36. ===========================可以看到对应不上, hystrix中复用的线程上下文也并未清除
  37. http-nio-9001-exec-2设置上下文=======>1652180438894
  38. http-nio-9001-exec-1设置上下文=======>1652180438894
  39. http-nio-9001-exec-3设置上下文=======>1652180438894
  40. http-nio-9001-exec-4设置上下文=======>1652180438964
  41. http-nio-9001-exec-5设置上下文=======>1652180439064
  42. hystrix-service-b-2获取上下文=======>1652180438894
  43. hystrix-service-b-1获取上下文=======>1652180439064
  44. hystrix-service-b-3获取上下文=======>1652180438964
  45. http-nio-9001-exec-6设置上下文=======>1652180439154
  46. http-nio-9001-exec-7设置上下文=======>1652180439254
  47. http-nio-9001-exec-8设置上下文=======>1652180439354
  48. http-nio-9001-exec-9设置上下文=======>1652180439454
  49. http-nio-9001-exec-10设置上下文=======>1652180439564
  50. hystrix-service-b-1获取上下文=======>1652180439064
  51. hystrix-service-b-2获取上下文=======>1652180438894
  52. hystrix-service-b-3获取上下文=======>1652180438964
  53. hystrix-service-b-2获取上下文=======>1652180438894
  54. hystrix-service-b-3获取上下文=======>1652180438964
  56. 4.再发起单次请求, 可以观察的更明显, hystrix获取的上下文为之前请求遗留的数据
  57. http-nio-9001-exec-1设置上下文=======>1652180593488
  58. hystrix-service-b-1获取上下文=======>1652180439064

方案一(推荐): 使用Hystrix插件机制, 用TtlCallable包装线程

  • 此方案的本质是针对HystrixCommand的run()方法（也就是加了@HystrixCommand注解的业务方法）拦截处理, 但它可能会超时或失败，那么就会去执行fallback方法，如果在 fallback方法中也想共享相关上下文信息，这时就无法覆盖到这种场景了

    1. @Slf4j
    2. @Configuration
    3. public class HystrixPluginConfiguration {
    5.   @PostConstruct
    6.   public void initHystrixPlugins() {
    7.       try {
    8.           HystrixConcurrencyStrategy target = new TtlHystrixConcurrencyStrategy();
    9.           HystrixConcurrencyStrategy strategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
    10.           if (strategy instanceof TtlHystrixConcurrencyStrategy) {
    11.               return;
    12.           }
    13.           HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins
    14.                   .getInstance().getCommandExecutionHook();
    15.           HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
    16.                   .getEventNotifier();
    17.           HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
    18.                   .getMetricsPublisher();
    19.           HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
    20.                   .getPropertiesStrategy();
    22.           if (log.isDebugEnabled()) {
    23.               log.debug("Current Hystrix plugins configuration is ["
    24.                       + "concurrencyStrategy [" + target + "]," + "eventNotifier ["
    25.                       + eventNotifier + "]," + "metricPublisher [" + metricsPublisher + "],"
    26.                       + "propertiesStrategy [" + propertiesStrategy + "]," + "]");
    27.               log.debug("Registering Ttl Hystrix Concurrency Strategy.");
    28.           }
    30.           HystrixPlugins.reset();
    31.           HystrixPlugins.getInstance().registerConcurrencyStrategy(target);
    32.           HystrixPlugins.getInstance()
    33.                   .registerCommandExecutionHook(commandExecutionHook);
    34.           HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
    35.           HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
    36.           HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
    37.       } catch (Exception e) {
    38.           log.error("Failed to register Ttl Hystrix Concurrency Strategy", e);
    39.       }
    40.   }
    42.   /**
    43.    * 使用TransmittableThreadLocal修饰Callable, 以实现线程池中上下文的正确传递
    44.    */
    45.   public static class TtlHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
    47.       @Override
    48.       public <T> Callable<T> wrapCallable(Callable<T> callable) {
    49.           return TtlCallable.get(callable);
    50.       }
    51.   }
    52. }

    测试: 同样以10个并发一次访问serviceA, 日志如下, 上下文一一对应 ```java http-nio-9001-exec-3设置上下文=======>1652180858093 http-nio-9001-exec-1设置上下文=======>1652180858093 http-nio-9001-exec-2设置上下文=======>1652180858173 http-nio-9001-exec-4设置上下文=======>1652180858273 hystrix-service-b-1获取上下文=======>1652180858093 hystrix-service-b-3获取上下文=======>1652180858093 hystrix-service-b-2获取上下文=======>1652180858173 http-nio-9001-exec-5设置上下文=======>1652180858376 http-nio-9001-exec-6设置上下文=======>1652180858472 http-nio-9001-exec-8设置上下文=======>1652180858566 http-nio-9001-exec-9设置上下文=======>1652180858673 http-nio-9001-exec-7设置上下文=======>1652180858773 hystrix-service-b-2获取上下文=======>1652180858376 hystrix-service-b-1获取上下文=======>1652180858472 hystrix-service-b-3获取上下文=======>1652180858273 hystrix-service-b-2获取上下文=======>1652180858673 hystrix-service-b-1获取上下文=======>1652180858566 hystrix-service-b-3获取上下文=======>1652180858773 http-nio-9001-exec-10设置上下文=======>1652180858883 hystrix-service-b-3获取上下文=======>1652180858883

再次单次请求, 观察得更清晰 http-nio-9001-exec-1设置上下文=======>1652180929774 hystrix-service-b-1获取上下文=======>1652180929774

  2. <a name="AG8IA"></a>
  3. ##### 方案二: 使用HystrixRequestContext上下文
  4. 参考: [https://www.freesion.com/article/9286656341/](https://www.freesion.com/article/9286656341/)<br />此方案只有使用HystrixContextRunnable或HystrixContextCallable创建线程才能在线程间传递数据, 在这里不过多介绍
  6. <a name="se3sK"></a>
  7. ## 业务案例
  8. <a name="kTwK3"></a>
  9. ### 编排异步任务实现并行召回
  10. 搜索业务有如下的并行召回流程, 下面我们用CompletableFuture模拟实现, 其中关键的mm, es, sm的并行召回<br />![search召回.png](https://cdn.nlark.com/yuque/0/2022/png/281275/1652258714747-af8799ed-eb0b-4347-b198-840600bec769.png#clientId=ua3bb9012-2db7-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=401&id=ue1133160&margin=%5Bobject%20Object%5D&name=search%E5%8F%AC%E5%9B%9E.png&originHeight=401&originWidth=771&originalType=binary&ratio=1&rotation=0&showTitle=false&size=22702&status=done&style=none&taskId=udcc1ca54-298f-4b39-bd1a-83a2690ddaa&title=&width=771)
  11. ```java
  12. /**
  13.  * Search
  14.  *
  15.  * @author xinzhang
  16.  * @version 2022/5/9
  17.  */
  18. @Slf4j
  19. @Component
  20. public class Search {
  22.     @Autowired
  23.     private ExecutorService ttlExecutor;
  25.     public List<Article> recall(String query) {
  26.         long start = System.currentTimeMillis();
  27.         // 设置UUID作为traceId
  28.         TransmittableThreadLocal<String> ctx = new TransmittableThreadLocal<>();
  29.         String traceId = UUID.fastUUID().toString();
  30.         System.out.println("=====================" + traceId + "===================");
  31.         ctx.set(traceId);
  33.         CompletableFuture<List<Article>> smRecall = CompletableFuture.supplyAsync(() -> {
  34.             // 子线程打印traceId
  35.             System.out.println(Thread.currentThread().getName()
  36.                     + "=====>smRecall executing, timeout<1s, ==>traceId: " + ctx.get());
  37.             randomSleepWithIn5Seconds(ctx.get());
  38.             return ArticleFactory.randomGenerate(query, "sm");
  39.         }, ttlExecutor);
  40.         // 设置超时以及异常处理
  41.         CompletableFuture<List<Article>> smRecallTimeOut = CompletableFutureTimeout
  42.                 .orTimeout(smRecall, 1, TimeUnit.SECONDS)
  43.                 .exceptionally(e -> {
  44.                     System.out.println(String.format("smRecall failed, e: %s", e.getMessage()));
  45.                     return Collections.emptyList();
  46.                 });
  48.         CompletableFuture<List<Article>> mmRecall = CompletableFuture.supplyAsync(() -> {
  49.             System.out.println(Thread.currentThread().getName()
  50.                     + "=====> mmRecall executing, timeout<2s, ==>traceId: " + ctx.get());
  51.             randomSleepWithIn5Seconds(ctx.get());
  52.             return ArticleFactory.randomGenerate(query, "mm");
  53.         }, ttlExecutor);
  54.         CompletableFuture<List<Article>> mmRecallTimeOut = CompletableFutureTimeout
  55.                 .orTimeout(mmRecall, 2, TimeUnit.SECONDS)
  56.                 .exceptionally(e -> {
  57.                     System.out.println(String.format("mmRecall failed, e: %s", e.getMessage()));
  58.                     return Collections.emptyList();
  59.                 });
  61.         CompletableFuture<List<Article>> esRecall = CompletableFuture.supplyAsync(() -> {
  62.             System.out.println(Thread.currentThread().getName()
  63.                     + "=====>esRecall executing, timeout<3s, ==>traceId: " + ctx.get());
  64.             randomSleepWithIn5Seconds(ctx.get());
  65.             return ArticleFactory.randomGenerate(query, "es");
  66.         }, ttlExecutor);
  67.         CompletableFuture<List<Article>> esRecallTimeOut = CompletableFutureTimeout
  68.                 .orTimeout(esRecall, 3, TimeUnit.SECONDS)
  69.                 .exceptionally(e -> {
  70.                     System.out.println(String.format("esRecall failed, e: %s", e.getMessage()));
  71.                     return Collections.emptyList();
  72.                 });
  74.         CompletableFuture<List<Article>> allDone = CompletableFuture
  75.                 .allOf(mmRecallTimeOut, esRecallTimeOut, smRecallTimeOut).thenApply(v -> {
  76.                     List<Article> all = new ArrayList<>();
  77.                     all.addAll(mmRecallTimeOut.join());
  78.                     all.addAll(esRecallTimeOut.join());
  79.                     all.addAll(smRecallTimeOut.join());
  80.                     return all;
  81.                 });
  82.         List<Article> list = allDone.join();
  83.         // 打印总耗时, 因为es, sm, mm超时时间最长为3s, 因此总耗时一定<=3s
  84.         System.out.println("recall total: " + (System.currentTimeMillis() - start) / 1000 + "s");
  85.         System.out.println(list);
  86.         return list;
  87.     }
  89.     /**
  90.      * 随机睡眠0~5s
  91.      */
  92.     public void randomSleepWithIn5Seconds(String traceId) {
  93.         Random random = new Random();
  94.         int time = random.nextInt(5000) - 1;
  95.         try {
  96.             System.out.println(Thread.currentThread().getName() + "睡眠" + time / 1000 + "s, ==>traceId: " + traceId);
  97.             Thread.sleep(time);
  98.         } catch (InterruptedException e) {
  99.             e.printStackTrace();
  100.         }
  101.     }
  102. }

可以看到, 超时处理以及异常处理等API的调用较为固定, 可以抽象出一个并行处理器来简化代码

  1. package com.pingan.lcloud.cf;
  3. import com.pingan.lcloud.jdk.CompletableFutureTimeout;
  4. import java.util.ArrayList;
  5. import java.util.List;
  6. import java.util.concurrent.CompletableFuture;
  7. import java.util.concurrent.ExecutorService;
  8. import java.util.concurrent.TimeUnit;
  9. import java.util.function.Function;
  10. import java.util.function.Supplier;
  11. import java.util.stream.Collectors;
  13. /**
  14.  * ParallelExecutor
  15.  * 并行执行器, 通过submit(), 提交任务, 可设置超时(超时会抛出TimeoutException)以及异常处理器,
  16.  * 通过execute()并行执行并获取结果
  17.  *
  18.  * @author xinzhang
  19.  * @version 2022/5/11
  20.  */
  21. public class ParallelExecutor<T> {
  23.     private final List<CompletableFuture<T>> tasks = new ArrayList<>();
  24.     private ExecutorService executorService;
  26.     public ParallelExecutor(ExecutorService executorService) {
  27.         this.executorService = executorService;
  28.     }
  30.     /**
  31.      * 任务提交, 超时或异常返回null
  32.      */
  33.     public void submit(Supplier<T> task, long timeout, TimeUnit unit) {
  34.         CompletableFuture<T> future = CompletableFuture.supplyAsync(task, executorService);
  35.         CompletableFuture<T> timeoutFuture = CompletableFutureTimeout.orTimeout(future, timeout, unit)
  36.                 .exceptionally(e -> null);
  37.         tasks.add(timeoutFuture);
  38.     }
  40.     /**
  41.      * 任务提交
  42.      *
  43.      * @param task 需要执行的任务
  44.      * @param timeout 超时时间
  45.      * @param unit 超时单位
  46.      * @param errorHandler 异常处理器
  47.      */
  48.     public void submit(Supplier<T> task, long timeout, TimeUnit unit, Function<Throwable, ? extends T> errorHandler) {
  49.         CompletableFuture<T> future = CompletableFuture.supplyAsync(task, executorService);
  50.         CompletableFuture<T> timeoutFuture = CompletableFutureTimeout.orTimeout(future, timeout, unit)
  51.                 .exceptionally(errorHandler);
  52.         tasks.add(timeoutFuture);
  53.     }
  55.     /**
  56.      * 任务执行
  57.      *
  58.      * @return 结果
  59.      */
  60.     public List<T> execute() {
  61.         return CompletableFuture.allOf(tasks.toArray(new CompletableFuture[]{}))
  62.                 .thenApply(v -> tasks.stream().map(CompletableFuture::join).collect(Collectors.toList())).join();
  63.     }
  64. }

使用

  1. /**
  2.  * Search
  3.  *
  4.  * @author xinzhang
  5.  * @version 2022/5/9
  6.  */
  7. @Slf4j
  8. @Component
  9. public class SearchV2 {
  11.     @Autowired
  12.     private ExecutorService ttlExecutor;
  14.     /**
  15.      * 召回错误处理器
  16.      */
  17.     static class RecallErrorHandler implements Function<Throwable, List<Article>> {
  19.         @Override
  20.         public List<Article> apply(Throwable t) {
  21.             System.out.println(String.format("recall failed, e: %s", t.getMessage()));
  22.             return Collections.emptyList();
  23.         }
  24.     }
  26.     public List<Article> recall(String query) {
  27.         long start = System.currentTimeMillis();
  28.         // 设置UUID作为traceId
  29.         TransmittableThreadLocal<String> ctx = new TransmittableThreadLocal<>();
  30.         String traceId = UUID.fastUUID().toString();
  31.         System.out.println("=====================" + traceId + "===================");
  32.         ctx.set(traceId);
  34.         // 定义错误处理器
  35.         RecallErrorHandler errorHandler = new RecallErrorHandler();
  37.         // 定义并行执行器
  38.         ParallelExecutor<List<Article>> parallelExecutor = new ParallelExecutor<>(ttlExecutor);
  39.         parallelExecutor.submit(() -> {
  40.             // 子线程打印traceId
  41.             System.out.println(Thread.currentThread().getName()
  42.                     + "=====>smRecall executing, timeout<1s, ==>traceId: " + ctx.get());
  43.             randomSleepWithIn5Seconds(ctx.get());
  44.             return ArticleFactory.randomGenerate(query, "sm");
  45.         }, 1, TimeUnit.SECONDS, errorHandler);
  47.         parallelExecutor.submit(() -> {
  48.             // 子线程打印traceId
  49.             System.out.println(Thread.currentThread().getName()
  50.                     + "=====>mmRecall executing, timeout<2s, ==>traceId: " + ctx.get());
  51.             randomSleepWithIn5Seconds(ctx.get());
  52.             return ArticleFactory.randomGenerate(query, "mm");
  53.         }, 2, TimeUnit.SECONDS, errorHandler);
  55.         parallelExecutor.submit(() -> {
  56.             // 子线程打印traceId
  57.             System.out.println(Thread.currentThread().getName()
  58.                     + "=====>esRecall executing, timeout<3s, ==>traceId: " + ctx.get());
  59.             randomSleepWithIn5Seconds(ctx.get());
  60.             return ArticleFactory.randomGenerate(query, "es");
  61.         }, 3, TimeUnit.SECONDS, errorHandler);
  63.         List<List<Article>> res = parallelExecutor.execute();
  64.         // 将list合并
  65.         List<Article> list = res.stream().flatMap(Collection::stream).collect(Collectors.toList());
  67.         // 打印总耗时, 因为es, sm, mm超时时间最长为3s, 因此总耗时一定<=3s
  68.         System.out.println("recall total: " + (System.currentTimeMillis() - start) / 1000 + "s");
  69.         System.out.println(list);
  70.         return list;
  71.     }
  73.     /**
  74.      * 随机睡眠0~5s
  75.      */
  76.     public void randomSleepWithIn5Seconds(String traceId) {
  77.         Random random = new Random();
  78.         int time = random.nextInt(5000) - 1;
  79.         try {
  80.             System.out.println(Thread.currentThread().getName() + "睡眠" + time / 1000 + "s, ==>traceId: " + traceId);
  81.             Thread.sleep(time);
  82.         } catch (InterruptedException e) {
  83.             e.printStackTrace();
  84.         }
  85.     }
  86. }