Распространение контекста Spring Boot 3 в трассировке микрометра

Question

Spring Boot 3 изменил распространение контекста в трассировке. https://github.com/micrometer-metrics/tracing/wiki/Spring-Cloud-Sleuth-3.1-Migration-Guide#async-instrumentation

Они поставляют сейчас библиотеку для этого выпуска. Наверное, я не совсем понимаю, как это работает. Я создал TaskExecutor, как в руководстве.

      @Bean(name = "taskExecutor")
    ThreadPoolTaskExecutor threadPoolTaskScheduler() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor() {
            @Override
            protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
                ExecutorService executorService = super.initializeExecutor(threadFactory, rejectedExecutionHandler);
                return ContextExecutorService.wrap(executorService, ContextSnapshot::captureAll);
            }
        };
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }

И я отметил @Async следующим образом:

       @Async("taskExecutor")
    public void run() {
        // invoke some service
    }

Но контекст не распространяется на дочерний контекст в потоке taskExecutor.

Answer 1

Вы можете автоматически подключить свойThreadPoolTaskExecutorи перенос контекста в AsyncConfigurer.

      import io.micrometer.context.ContextExecutorService;
import io.micrometer.context.ContextSnapshot;
import java.util.concurrent.Executor;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration(proxyBeanMethods = false)
@RequiredArgsConstructor
public class AsyncTraceContextConfig implements AsyncConfigurer {
  
  // NOTE: By design you can only have one AsyncConfigurer, thus only one executor pool is
  // configurable.
  @Qualifier("taskExecutor") // if you have more than one task executor pools
  private final ThreadPoolTaskExecutor taskExecutor;

  @Override
  public Executor getAsyncExecutor() {
    return ContextExecutorService.wrap(
        taskExecutor.getThreadPoolExecutor(), ContextSnapshot::captureAll);
  }
}

ОБНОВЛЯТЬ

Если у вас несколько пулов исполнителей и вы хотите добавить трассировку для всех, используйте командуTaskDecoratorсContextSnapshot.wrap():

      import io.micrometer.context.ContextSnapshot;
import java.util.concurrent.Executor;
import org.springframework.boot.task.TaskExecutorBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskDecorator;

@Configuration
public class AsyncConfig {
  @Bean
  public TaskDecorator otelTaskDecorator() {
    return (runnable) -> ContextSnapshot.captureAll(new Object[0]).wrap(runnable);
  }

  @Bean("asyncExecutorPool1")
  public Executor asyncExecutorPool1(TaskDecorator otelTaskDecorator) {
    return new TaskExecutorBuilder()
        .corePoolSize(5)
        .maxPoolSize(10)
        .queueCapacity(10)
        .threadNamePrefix("threadPoolExecutor1-")
        .taskDecorator(otelTaskDecorator)
        .build();
  }

  @Bean("asyncExecutorPool2")
  public Executor asyncExecutorPool2(TaskDecorator otelTaskDecorator) {
    return new TaskExecutorBuilder()
        .corePoolSize(5)
        .maxPoolSize(10)
        .queueCapacity(10)
        .threadNamePrefix("threadPoolExecutor2-")
        .taskDecorator(otelTaskDecorator)
        .build();
  }
}

ПРИМЕЧАНИЕ. Вы можете следить за этим блогом , чтобы получить более подробную информацию о настройке и пример кода проекта GitHub.

Answer 2

Я столкнулся с той же проблемой. Пожалуйста, добавьте этот код в конфигурацию, и все будет работать как положено.

        @Configuration(proxyBeanMethods = false)
  static class AsyncConfig implements AsyncConfigurer, WebMvcConfigurer {

    @Override
    public Executor getAsyncExecutor() {
      return ContextExecutorService.wrap(Executors.newCachedThreadPool(), ContextSnapshot::captureAll);
    }

    @Override
    public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
      configurer.setTaskExecutor(new SimpleAsyncTaskExecutor(r -> new Thread(ContextSnapshot.captureAll().wrap(r))));
    }
  }

Answer 3

Вы можете попробовать и так, но я подозреваю, что это ошибка.

      ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(){
            @Override
            protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
                ExecutorService executorService = super.initializeExecutor(threadFactory, rejectedExecutionHandler);
                return ContextExecutorService.wrap(executorService, ContextSnapshot::captureAll);
            }
            @Override
            public void execute(Runnable task) {
                super.execute(ContextSnapshot.captureAll().wrap(task));
            }

            @Override
            public Future<?> submit(Runnable task) {
                return super.submit(ContextSnapshot.captureAll().wrap(task));
            }

            @Override
            public <T> Future<T> submit(Callable<T> task) {
                return super.submit(ContextSnapshot.captureAll().wrap(task));
            }
        };

Answer 4

Зарегистрируйте компонент ContextPropagatingTaskDecorator. Он будет выбран при автоматической настройке и подключен к ThreadPoolTaskExecutorBuilder.

См.: org.springframework.boot.autoconfigure.task.TaskExecutorConfigurations.ThreadPoolTaskExecutorBuilderConfiguration#threadPoolTaskExecutorBuilder.

         @Bean
   public TaskDecorator decorator(){
       return new ContextPropagatingTaskDecorator();
   }

Таким образом, вы можете использовать его в

      
   @Autowired
   ThreadPoolTaskExecutor scheduler;


   void test() {
        scheduler.submit(() -> log.info("Running task using scheduler "));
   }