Java線程之間數(shù)據(jù)傳遞的實現(xiàn)示例(4種)

更新時間：2023年08月06日 09:00:48 作者：立瑩Sir

我們經(jīng)常會遇到父子線程數(shù)據(jù)傳遞(非調(diào)用參數(shù))的場景,本文主要介紹了Java線程之間數(shù)據(jù)傳遞的實現(xiàn)示例,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

前言

在業(yè)務(wù)系統(tǒng)的開發(fā)過程中，我們經(jīng)常會遇到父子線程數(shù)據(jù)傳遞(非調(diào)用參數(shù))的場景，如：登陸信息，調(diào)用者信息，TraceId的傳遞等業(yè)務(wù)場景，固總結(jié)4中方式進行線程之間數(shù)據(jù)傳遞。

ThreadLocal

代碼如下：

public class TtlParameterWrapper {
? ? private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<>();
? ? private TtlParameterWrapper() {
? ? }
? ? public static String getCaller() {
? ? ? ? return THREAD_LOCAL.get();
? ? }
? ? public static void setCaller(String caller) {
? ? ? ?THREAD_LOCAL.set(caller);
? ? }
? ? public static void clear() {
? ? ? ?THREAD_LOCAL.remove();
? ? }
}

那么子線程想要獲取這個TtlParameterWrapper如何做呢？

獲取父線程的TtlParameterWrapper
將TtlParameterWrapper設(shè)置到子線程，達到復(fù)用

public void handler(){
        // 1. 獲取父線程
       TtlParameterWrapper.setCaller("caller path");
       log.info("父線程的值 ->{}",TtlParameterWrapper.get());
       CompletableFuture.runAsync(()->{
            // 2. 設(shè)置子線程的值，復(fù)用
          TtlParameterWrapper.setCaller("caller path");
          log.info("子線程的值 ->{}", TtlParameterWrapper.getCaller());
        });
    }

總結(jié)

雖然最終達成了傳遞的目的，但是每次開異步線程都需要手動設(shè)置，代碼冗余繁雜，如果不這樣設(shè)置則無法跨線程進行傳遞；如果手動設(shè)置，將無法進行線程間進行傳遞，因為TheadLocal中的數(shù)據(jù)無法進行線程間進行傳遞。

InheritableThreadLocal

這種方案不建議使用，InheritableThreadLocal雖然能夠?qū)崿F(xiàn)父子線程間的復(fù)用，但是在線程池中使用會存在失敗的問題，原因：InheritableThreadLocal 在父線程創(chuàng)建子線程的時候，會將父線程中InheritableThreadLocal中存儲的數(shù)據(jù) 拷貝一份存儲到子線程的 InheritableThreadLocal中，但是在web的容器中使用了線程池，線程會被創(chuàng)建回收重復(fù)的利用，不會被銷毀重新創(chuàng)建，所以會存在實效的場景。

這種方案使用也是非常簡單，直接用InheritableThreadLocal替換ThreadLocal即可。

代碼如下：

public class TtlParameterWrapper {
? ? private static ?final ?InheritableThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>();
? ? public static String getCaller(){
? ? ? ? return inheritableThreadLocal.get();
? ? }
? ? public static void setCaller(LoginVal loginVal){
? ? ? ?inheritableThreadLocal.set(loginVal);
? ? }
? ? public static void clear(){
? ? ? ?inheritableThreadLocal.remove();
? ? }
}

TransmittableThreadLocal

TransmittableThreadLocal是阿里開源的工具，解決了InheritableThreadLocal不能進行線程池間傳遞數(shù)據(jù)的缺陷，在使用線程池等會池化復(fù)用線程的執(zhí)行組件情況下，提供ThreadLocal值的傳遞功能，解決異步執(zhí)行時上下文傳遞的問題。

添加依賴

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>transmittable-thread-local</artifactId>
    <version>2.14.2</version>
</dependency>

TtlParameterWrapper改造

public class TtlParameterWrapper {
    private static final TransmittableThreadLocal<String> TRANSMITTABLE_THREAD_LOCAL = new TransmittableThreadLocal<>();
    private TtlParameterWrapper() {
    }
    public static String getCaller() {
        return TRANSMITTABLE_THREAD_LOCAL.get();
    }
    public static void setCaller(String caller) {
       TRANSMITTABLE_THREAD_LOCAL.set(caller);
    }
    public static void clear() {
        TRANSMITTABLE_THREAD_LOCAL.remove();
    }
}

原理

從定義來看，TransimittableThreadLocal繼承于InheritableThreadLocal，并實現(xiàn)TtlCopier接口，它里面只有一個copy方法。所以主要是對InheritableThreadLocal的擴展。

public class TransmittableThreadLocal<T> extends InheritableThreadLocal<T> implements TtlCopier<T>

在TransimittableThreadLocal中添加holder屬性。這個屬性的作用就是被標記為具備線程傳遞資格的對象都會被添加到這個對象中。
要標記一個類，比較容易想到的方式，就是給這個類新增一個Type字段，還有一個方法就是將具備這種類型的的對象都添加到一個靜態(tài)全局集合中。之后使用時，這個集合里的所有值都具備這個標記。

//1. holder本身是一個InheritableThreadLocal對象
//2. 這個holder對象的value是WeakHashMap<TransmittableThreadLocal<Object>,?>
// ? 2.1WeekHashMap的value總是null,且不可能被使用。
// ? ?2.2WeekHasshMap支持value=null
private static InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>,?>> holder = new InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>,?>>(){
@Override
protected WeakHashMap<TransmittableThreadLocal<Object>, ?> initialValue() {
?? ?return new WeakHashMap<TransmittableThreadLocal<Object>, Object>();
}
/**
?* 重寫了childValue方法，實現(xiàn)上直接將父線程的屬性作為子線程的本地變量對象。 ??
?*/
@Override
protected WeakHashMap<TransmittableThreadLocal<Object>, ?> childValue(WeakHashMap<TransmittableThreadLocal<Object>, ?>parentValue){
?? ?return new WeakHashMap<TransmittableThreadLocal<Object>, Object>(parentValue);
}};

應(yīng)用代碼是通過TtlExecutors工具類對線程池對象進行包裝。工具類只是簡單的判斷，輸入的線程池是否已經(jīng)被包裝過、非空校驗等，然后返回包裝類ExecutorServiceTtlWrapper。根據(jù)不同的線程池類型，有不同和的包裝類。

@Nullable
public static ExecutorServicegetTtlExecutorService(@Nullable ExecutorService executorService){
    if (TtlAgent.isTtlAgentLoaded() || executorService == null || executorService instanceof TtlEnhanced){
        return executorService;
    }
    return new ExecutorServiceTtlWrapper(executorService);
}

進入包裝類ExecutorServiceTtlWrapper?？梢宰⒁獾讲徽撌峭ㄟ^ExecutorServiceTtlWrapper#submit方法或者是ExecutorTtlWrapper#execute方法，都會將線程對象包裝成TtlCallable或者TtlRunnable，用于在真正執(zhí)行run方法前做一些業(yè)務(wù)邏輯。

/** 
 * 在ExecutorServiceTtlWrapper實現(xiàn)submit方法
 */
@NonNull
@Override
public <T> Future<T> submit(@NonNull Callable<T> task){
    return executorService.submit(TtlCallable.get(task));
}
/** 
 * 在ExecutorTtlWrapper實現(xiàn)execute方法
 */
@Override
public void execute(@NonNull Runnable command){
    executor.execute(TtlRunnable.get(command));
}

重點的核心邏輯應(yīng)該是在TtlCallable#call()或者TtlRunnable#run()中。以下以TtlCallable為例，TtlRunnable同理類似。在分析call()方法之前，先看一個類Transmitter。

public static class Transmitter {
    /**    
     * 捕獲當前線程中的是所有TransimittableThreadLocal和注冊ThreadLocal的值。    
     */
    @NonNull
    public static Object capture(){
        return new Snapshot(captureTtlValues(), captureThreadLocalValues());
    }
    /**    
     * 捕獲TransimittableThreadLocal的值,將holder中的所有值都添加到HashMap后返回。    
     */
    private static HashMap<TransmittableThreadLocal<Object>, Object> captureTtlValues() {
        HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value = new HashMap<TransmittableThreadLocal<Object>, Object>();
        for(TransmittableThreadLocal<Object> threadLocal : holder.get().keySet()) {
            ttl2Value.put(threadLocal, threadLocal.copyValue());
        }
        return ttl2Value;
    }
    /**    
     * 捕獲注冊的ThreadLocal的值,也就是原本線程中的ThreadLocal,可以注冊到TTL中，在    
     * 進行線程池本地變量傳遞時也會被傳遞。   
     */
    private static HashMap<ThreadLocal<Object>, Object> captureThreadLocalValues() {
        final HashMap<ThreadLocal<Object>, Object> threadLocal2Value = new HashMap<ThreadLocal<Object>, Object>();
        for(Map.Entry<ThreadLocal<Object>, TtlCopier<Object>> entry : threadLocalHolder.entrySet()) {
            final ThreadLocal<Object> threadLocal = entry.getKey();
            final TtlCopier<Object> copier = entry.getValue();
            threadLocal2Value.put(threadLocal, copier.copy(threadLocal.get()));
        }
        return threadLocal2Value;
    }
    /**    
     * 將捕獲到的本地變量進行替換子線程的本地變量，并且返回子線程現(xiàn)有的本地變量副本backup。   
     * 用于在執(zhí)行run/call方法之后，將本地變量副本恢復(fù)。    
     */
    @NonNull
    public static Object replay(@NonNull Object captured) {
        final Snapshot capturedSnapshot = (Snapshot) captured;
        return new Snapshot(replayTtlValues(capturedSnapshot.ttl2Value), replayThreadLocalValues(capturedSnapshot.threadLocal2Value));
    }
    /**    
     * 替換TransmittableThreadLocal    
     */
    @NonNull
    private static HashMap<TransmittableThreadLocal<Object>, Object> replayTtlValues(@NonNull HashMap<TransmittableThreadLocal<Object>,Object> captured) {
        // 創(chuàng)建副本backup
        HashMap<TransmittableThreadLocal<Object>, Object> backup = new HashMap<TransmittableThreadLocal<Object>, Object>();
        for (final Iterator<TransmittableThreadLocal<Object>> iterator = holder.get().keySet().iterator(); iterator.hasNext();) {
            TransmittableThreadLocal<Object> threadLocal = iterator.next();
            // 對當前線程的本地變量進行副本拷貝
            backup.put(threadLocal, threadLocal.get());
            // 若出現(xiàn)調(diào)用線程中不存在某個線程變量，而線程池中線程有，則刪除線程池中對應(yīng)的本地變量
            if (!captured.containsKey(threadLocal)){
                iterator.remove();
                threadLocal.superRemove();
            }
        }
        // 將捕獲的TTL值打入線程池獲取到的線程TTL中。
        setTtlValuesTo(captured);
        // 是一個擴展點，調(diào)用TTL的beforeExecute方法。默認實現(xiàn)為空
        doExecuteCallback(true);
        return backup;
    }
    private static HashMap<ThreadLocal<Object>, Object> replayThreadLocalValues(@NonNull HashMap<ThreadLocal<Object>, Object> captured) {
        final HashMap<ThreadLocal<Object>, Object> backup = new HashMap<ThreadLocal<Object>, Object>();
        for (Map.Entry<ThreadLocal<Object>, Object> entry : captured.entrySet()) {
            final ThreadLocal<Object> threadLocal = entry.getKey();
            backup.put(threadLocal, threadLocal.get());
            final Objectvalue = entry.getValue();
            if (value == threadLocalClearMark) threadLocal.remove();
            else threadLocal.set(value);
        }
        return backup;
    }
    /**    
     * 清除單線線程的所有TTL和TL，并返回清除之氣的backup    
     */
    @NonNull
    public static Object clear() {
        final HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value = new HashMap<TransmittableThreadLocal<Object>, Object>();
        final HashMap<ThreadLocal<Object>, Object> threadLocal2Value = new HashMap<ThreadLocal<Object>, Object>();
        for (Map.Entry<ThreadLocal<Object>,TtlCopier<Object>>entry : threadLocalHolder.entrySet()) {
            final ThreadLocal<Object> threadLocal = entry.getKey();
            threadLocal2Value.put(threadLocal, threadLocalClearMark);
        }
        return replay(new Snapshot(ttl2Value, threadLocal2Value));
    }
    /**    
     * 還原    
     */
    public static void restore(@NonNull Object backup) {
        final Snapshot backupSnapshot = (Snapshot) backup;
        restoreTtlValues(backupSnapshot.ttl2Value);
        restoreThreadLocalValues(backupSnapshot.threadLocal2Value);
    }
    private static void restoreTtlValues(@NonNull HashMap<TransmittableThreadLocal<Object>,Object> backup) {
        // 擴展點，調(diào)用TTL的afterExecute
        doExecuteCallback(false);
        for (finalIterator<TransmittableThreadLocal<Object>> iterator = holder.get().keySet().iterator(); iterator.hasNext();) {
            TransmittableThreadLocal<Object> threadLocal = iterator.next();
            if (!backup.containsKey(threadLocal)) {
                iterator.remove();
                threadLocal.superRemove();
            }
        }
        // 將本地變量恢復(fù)成備份版本
        setTtlValuesTo(backup);
    }
    private static void setTtlValuesTo(@NonNull HashMap<TransmittableThreadLocal<Object>,Object> ttlValues) {
        for (Map.Entry<TransmittableThreadLocal<Object>, Object> entry :ttlValues.entrySet()) {
            TransmittableThreadLocal<Object> threadLocal = entry.getKey();
            threadLocal.set(entry.getValue());
        }
    }
    private staticvoid restoreThreadLocalValues(@NonNull HashMap<ThreadLocal<Object>,Object> backup) {
        for (Map.Entry<ThreadLocal<Object>, Object> entry : backup.entrySet()) {
            final ThreadLocal<Object> threadLocal = entry.getKey();
            threadLocal.set(entry.getValue());
        }
    }
    /**   
     * 快照類，保存TTL和TL   
     */
    private static class Snapshot {
        final HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value;
        final HashMap<ThreadLocal<Object>, Object> threadLocal2Value;
        private Snapshot(HashMap<TransmittableThreadLocal<Object>, Object>ttl2Value, HashMap<ThreadLocal<Object>, Object> threadLocal2Value) {
            this.ttl2Value= ttl2Value;
            this.threadLocal2Value= threadLocal2Value;
        }
    }
    // 進入TtlCallable#call()方法。
    @Override
    public V call() throws Exception {
        Object captured = capturedRef.get();
        if (captured == null|| releaseTtlValueReferenceAfterCall && !capturedRef.compareAndSet(captured, null)) {
            throw new IllegalStateException("TTLvalue reference is released after call!");
        }
        // 調(diào)用replay方法將捕獲到的當前線程的本地變量，傳遞給線程池線程的本地變量，
        // 并且獲取到線程池線程覆蓋之前的本地變量副本。
        Object backup = replay(captured);
        try {
            // 線程方法調(diào)用
            return callable.call();
        } finally {
            // 使用副本進行恢復(fù)。
            restore(backup);
        }
    }
}

線程池方式傳遞本地變量的核心代碼已經(jīng)完畢?？偟膩碚f在創(chuàng)建TtlCallable對象是，調(diào)用capture()方法捕獲調(diào)用方的本地線程變量，在call()執(zhí)行時，將捕獲到的線程變量，替換到線程池所對應(yīng)獲取到的線程的本地變量中，并且在執(zhí)行完成之后，將其本地變量恢復(fù)到調(diào)用之前。

TaskDecorator

線程池設(shè)置TaskDecorator，TaskDecorator是什么？

官方釋義：這是一個執(zhí)行回調(diào)方法的裝飾器，主要應(yīng)用于傳遞上下文，或者提供任務(wù)的監(jiān)控/統(tǒng)計信息。

代碼如下

public class ContextTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable){
        //獲取父線程的值
        String callerPath = TtlParameterWrapper.getCaller();
        return () -> {
            try {
                // 將主線程的請求信息，設(shè)置到子線程中
               TtlParameterWrapper.setCaller(callerPath);
                // 執(zhí)行子線程，這一步不要忘了
               runnable.run();
            } finally {
                // 線程結(jié)束，清空這些信息，否則可能造成內(nèi)存泄漏
               TtlParameterWrapper.clear();
            }
        };
    }
}

TaskDecorator需要結(jié)合線程池使用，實際開發(fā)中異步線程建議使用線程池，只需要在對應(yīng)的線程池配置一下

代碼

@Bean("taskExecutor")
public ThreadPoolTaskExecutor taskExecutor() {
       ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
       poolTaskExecutor.setCorePoolSize(xx);
       poolTaskExecutor.setMaxPoolSize(xx);
        // 設(shè)置線程活躍時間（秒）
       poolTaskExecutor.setKeepAliveSeconds(xx);
        // 設(shè)置隊列容量
       poolTaskExecutor.setQueueCapacity(xx);
        //設(shè)置TaskDecorator，用于解決父子線程間的數(shù)據(jù)復(fù)用
       poolTaskExecutor.setTaskDecorator(new ContextTaskDecorator());
       poolTaskExecutor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());
        // 等待所有任務(wù)結(jié)束后再關(guān)閉線程池
       poolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
       return poolTaskExecutor;
}

此時業(yè)務(wù)代碼就不需要去設(shè)置子線程的值，直接使用即可

代碼

public void handlerAsync() {
       log.info("父線程的用戶信息 -> {}", TtlParameterWrapper.get());
        //執(zhí)行異步任務(wù)，需要指定的線程池
       CompletableFuture.runAsync(() -> 
           log.info("子線程的用戶信息 -> {}", TtlParameterWrapper.get()
       ),taskExecutor);
}

這里使用的是CompletableFuture執(zhí)行異步任務(wù)，使用@Async這個注解同樣是可行的。

注意：無論使用何種方式，都需要指定線程池

到此這篇關(guān)于Java線程之間數(shù)據(jù)傳遞的實現(xiàn)示例的文章就介紹到這了,更多相關(guān)Java線程之間數(shù)據(jù)傳遞內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: