How to implement the transmission of traceId asynchronous thread

Preface

In daily problem investigation, we often query the requested log link based on traceId in ELK. In synchronous request, it is very cool to stop by traceId. So how should we deal with it if it is an asynchronous request?

Asynchronous requests in the project are used to start asynchronous threads in combination with thread pools. The following is a combination of MDC and thread pools in slf4j to realize traceId delivery of asynchronous threads.

Rewrite the method in ThreadPoolTaskExecutor

The following tool classes set the request mapping context through (context) before the Callable and Runnable asynchronous tasks are executed.

import org.;
import ;

import ;
import ;

/**
  * @desc: Define MDC tool class, supports Runnable and Callable, the purpose is to set the traceId of the parent thread to the child thread
  */
public class MdcUtil {

    public static &lt;T&gt; Callable&lt;T&gt; wrap(final Callable&lt;T&gt; callable, final Map&lt;String, String&gt; context) {
        return () -&gt; {
            if ((context)) {
                ();
            } else {
                (context);
            }
            try {
                return ();
            } finally {
                // Clear child threads to avoid memory overflow, which is the same as ()                ();
            }
        };
    }

    public static Runnable wrap(final Runnable runnable, final Map&lt;String, String&gt; context) {
        return () -&gt; {
            if ((context)) {
                ();
            } else {
                (context);
            }
            try {
                ();
            } finally {
                ();
            }
        };
    }
}

The following defines a ThreadPoolMdcExecutor class to inherit the ThreadPoolTaskExecutor class and override the execute and submit methods

import ;
import ;

/**
  * @desc: Pass the current traceId to the implementation specially added by the child thread.
  * The point is (), this method gets the traceId of the current thread (parent thread)
  */
public class ThreadPoolMdcExecutor extends ThreadPoolTaskExecutor {
    @Override
    public void execute(Runnable task) {
        ((task, ()));
    }

    @Override
    public Future&lt;?&gt; submit(Runnable task) {
        return ((task, ()));
    }

    @Override
    public &lt;T&gt; Future&lt;T&gt; submit(Callable&lt;T&gt; task) {
        return ((task, ()));
    }
}

Define the thread pool below, you can use ThreadPoolMdcExecutor

    @Bean(name = "callBackExecutorConfig")
    public Executor callBackExecutorConfig() {
        ThreadPoolTaskExecutor executor = new ThreadPoolMdcExecutor();
        // Configure the number of core threads        (10);
        // Configure the maximum number of threads        (20);
        // Configure queue size        (200);
        // Configure the prefix of the thread name in the thread pool        ("async-Thread-");
        // rejection-policy: How to handle new tasks when the pool has reached max size        // abort: Throw an exception in the method that calls the executor to execute RejectedExecutionException        (new ());
        // Execute initialization        ();
        return executor;
    }

After defining the thread pool, we can use the callBackExecutorConfig thread pool to perform asynchronous tasks to avoid the loss of traceId in the asynchronous thread.

Thread pool enhancement

The above is to inherit ThreadPoolTaskExecutor, override the execute and submit methods, and set the (context) setting context. We can also enhance the thread pool by implementing the TaskDecorator interface.

public class ContextTransferTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        Map<String, String> context = ();
        RequestAttributes requestAttributes = ();
        return () -> {
            try {
                (context);
                (requestAttributes);
                ();
            } finally {
                ();
                ();
            }
        };
    }
}

Next, define the thread pool and enhance the thread pool

    @Bean(name = "callBackExecutorConfig")
    public Executor callBackExecutorConfig() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor ();
        // Configure the number of core threads        (10);
        // Configure the maximum number of threads        (20);
        // Configure queue size        (200);
        // Configure the prefix of the thread name in the thread pool        ("async-Thread-");
        // rejection-policy: How to handle new tasks when the pool has reached max size        // abort: Throw an exception in the method that calls the executor to execute RejectedExecutionException        (new ());
        //Thread pool enhancement        (new ContextTransferTaskDecorator());
        // Execute initialization        ();
        return executor;
    }

Summarize

The above two methods are essentially synchronized by Mdc between asynchronous threads. You can look at the source code of Mdc, and finally, through InheritableThreadLocal to obtain parent thread information.

public class BasicMDCAdapter implements MDCAdapter {
    private InheritableThreadLocal&lt;Map&lt;String, String&gt;&gt; inheritableThreadLocal = 
        new InheritableThreadLocal&lt;Map&lt;String, String&gt;&gt;() {
        protected Map&lt;String, String&gt; childValue(Map&lt;String, String&gt; parentValue) {
            return parentValue == null ? null : new HashMap(parentValue);
        }
    };
    
    //Omit a few    ......
}

The above is personal experience. I hope you can give you a reference and I hope you can support me more.