如何在 Spring 異步調(diào)用中傳遞上下文

image

什么是異步調(diào)用？

異步調(diào)用是相對于同步調(diào)用而言的，同步調(diào)用是指程序按預(yù)定順序一步步執(zhí)行，每一步必須等到上一步執(zhí)行完后才能執(zhí)行，異步調(diào)用則無需等待上一步程序執(zhí)行完即可執(zhí)行。異步調(diào)用指，在程序在執(zhí)行時，無需等待執(zhí)行的返回值即可繼續(xù)執(zhí)行后面的代碼。在我們的應(yīng)用服務(wù)中，有很多業(yè)務(wù)邏輯的執(zhí)行操作不需要同步返回（如發(fā)送郵件、冗余數(shù)據(jù)表等），只需要異步執(zhí)行即可。

本文將介紹 Spring 應(yīng)用中，如何實現(xiàn)異步調(diào)用。在異步調(diào)用的過程中，會出現(xiàn)線程上下文信息的丟失，我們該如何解決線程上下文信息的傳遞。

Spring 應(yīng)用中實現(xiàn)異步

Spring 為任務(wù)調(diào)度與異步方法執(zhí)行提供了注解支持。通過在方法或類上設(shè)置 @Async 注解，可使得方法被異步調(diào)用。調(diào)用者會在調(diào)用時立即返回，而被調(diào)用方法的實際執(zhí)行是交給 Spring 的 TaskExecutor 來完成的。所以被注解的方法被調(diào)用的時候，會在新的線程中執(zhí)行，而調(diào)用它的方法會在原線程中執(zhí)行，這樣可以避免阻塞，以及保證任務(wù)的實時性。

引入依賴

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

引入 Spring 相關(guān)的依賴即可。

入口類

@SpringBootApplication
@EnableAsync
public class AsyncApplication {
    public static void main(String[] args) {
        SpringApplication.run(AsyncApplication.class, args);
    }

入口類增加了 @EnableAsync 注解，主要是為了掃描范圍包下的所有 @Async 注解。

對外的接口

這里寫了一個簡單的接口：

@RestController
@Slf4j
public class TaskController {

    @Autowired
    private TaskService taskService;

    @GetMapping("/task")
    public String taskExecute() {
        try {
            taskService.doTaskOne();
            taskService.doTaskTwo();
            taskService.doTaskThree();
        } catch (Exception e) {
           log.error("error executing task for {}",e.getMessage());
        }
        return "ok";
    }
}

調(diào)用 TaskService 執(zhí)行三個異步方法。

Service 方法

@Component
@Slf4j
//@Async
public class TaskService {

    @Async
    public void doTaskOne() throws Exception {
        log.info("開始做任務(wù)一");
        long start = System.currentTimeMillis();
        Thread.sleep(1000);
        long end = System.currentTimeMillis();
        log.info("完成任務(wù)一，耗時：" + (end - start) + "毫秒");
    }

    @Async
    public void doTaskTwo() throws Exception {
        log.info("開始做任務(wù)二");
        long start = System.currentTimeMillis();
        Thread.sleep(1000);
        long end = System.currentTimeMillis();
        log.info("完成任務(wù)二，耗時：" + (end - start) + "毫秒");
    }

    @Async
    public void doTaskThree() throws Exception {
        log.info("開始做任務(wù)三");
        long start = System.currentTimeMillis();
        Thread.sleep(1000);
        long end = System.currentTimeMillis();
        log.info("完成任務(wù)三，耗時：" + (end - start) + "毫秒");
    }
}

@Async 可以用于類上，標識該類的所有方法都是異步方法，也可以單獨用于某些方法。每個方法都會 sleep 1000 ms。

結(jié)果展示

運行結(jié)果如下：

image

可以看到 TaskService 中的三個方法是異步執(zhí)行的，接口的結(jié)果快速返回，日志信息異步輸出。異步調(diào)用，通過開啟新的線程調(diào)用的方法，不影響主線程。異步方法實際的執(zhí)行交給了 Spring 的 TaskExecutor 來完成。

Future：獲取異步執(zhí)行的結(jié)果

在上面的測試中我們也可以發(fā)現(xiàn)主調(diào)用方法并沒有等到調(diào)用方法執(zhí)行完就結(jié)束了當(dāng)前的任務(wù)。如果想要知道調(diào)用的三個方法全部執(zhí)行完該怎么辦呢，下面就可以用到異步回調(diào)。

異步回調(diào)就是讓每個被調(diào)用的方法返回一個 Future 類型的值，Spring 中提供了一個 Future 接口的子類：AsyncResult，所以我們可以返回 AsyncResult 類型的值。

public class AsyncResult<V> implements ListenableFuture<V> {

    private final V value;

    private final ExecutionException executionException;
    //...
}

AsyncResult 實現(xiàn)了 ListenableFuture 接口，該對象內(nèi)部有兩個屬性：返回值和異常信息。

public interface ListenableFuture<T> extends Future<T> {
    void addCallback(ListenableFutureCallback<? super T> var1);

    void addCallback(SuccessCallback<? super T> var1, FailureCallback var2);
}

ListenableFuture 接口繼承自 Future，在此基礎(chǔ)上增加了回調(diào)方法的定義。Future 接口定義如下：

public interface Future<V> {
    // 是否可以打斷當(dāng)前正在執(zhí)行的任務(wù)
    boolean cancel(boolean mayInterruptIfRunning);
    
    // 任務(wù)取消的結(jié)果
    boolean isCancelled();
    
    // 異步方法中最后返回的那個對象中的值 
    V get() throws InterruptedException, ExecutionException;
    // 用來判斷該異步任務(wù)是否執(zhí)行完成，如果執(zhí)行完成，則返回 true，如果未執(zhí)行完成，則返回false
    boolean isDone();
    // 與 get() 一樣，只不過這里參數(shù)中設(shè)置了超時時間
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

#get() 方法，在執(zhí)行的時候是需要等待回調(diào)結(jié)果的，阻塞等待。如果不設(shè)置超時時間，它就阻塞在那里直到有了任務(wù)執(zhí)行完成。我們設(shè)置超時時間，就可以在當(dāng)前任務(wù)執(zhí)行太久的情況下中斷當(dāng)前任務(wù)，釋放線程，這樣就不會導(dǎo)致一直占用資源。

#cancel(boolean) 方法，參數(shù)是一個 boolean 類型的值，用來傳入是否可以打斷當(dāng)前正在執(zhí)行的任務(wù)。如果參數(shù)是 true 且當(dāng)前任務(wù)沒有執(zhí)行完成 ，說明可以打斷當(dāng)前任務(wù)，那么就會返回 true；如果當(dāng)前任務(wù)還沒有執(zhí)行，那么不管參數(shù)是 true 還是 false，返回值都是 true；如果當(dāng)前任務(wù)已經(jīng)完成，那么不管參數(shù)是 true 還是 false，那么返回值都是 false；如果當(dāng)前任務(wù)沒有完成且參數(shù)是 false，那么返回值也是 false。即：

  1. 如果任務(wù)還沒執(zhí)行，那么如果想取消任務(wù)，就一定返回 true，與參數(shù)無關(guān)。
  2. 如果任務(wù)已經(jīng)執(zhí)行完成，那么任務(wù)一定是不能取消的，所以此時返回值都是false，與參數(shù)無關(guān)。
  3. 如果任務(wù)正在執(zhí)行中，那么此時是否取消任務(wù)就看參數(shù)是否允許打斷（true/false）。

獲取異步方法返回值的實現(xiàn)

    public Future<String> doTaskOne() throws Exception {
        log.info("開始做任務(wù)一");
        long start = System.currentTimeMillis();
        Thread.sleep(1000);
        long end = System.currentTimeMillis();
        log.info("完成任務(wù)一，耗時：" + (end - start) + "毫秒");
        return new AsyncResult<>("任務(wù)一完成，耗時" + (end - start) + "毫秒");
    }
    //...其他兩個方法類似，省略

我們將 task 方法的返回值改為 Future<String>，將執(zhí)行的時間拼接為字符串返回。

    @GetMapping("/task")
    public String taskExecute() {
        try {
            Future<String> r1 = taskService.doTaskOne();
            Future<String> r2 = taskService.doTaskTwo();
            Future<String> r3 = taskService.doTaskThree();
            while (true) {
                if (r1.isDone() && r2.isDone() && r3.isDone()) {
                    log.info("execute all tasks");
                    break;
                }
                Thread.sleep(200);
            }
            log.info("\n" + r1.get() + "\n" + r2.get() + "\n" + r3.get());
        } catch (Exception e) {
           log.error("error executing task for {}",e.getMessage());
        }

        return "ok";
    }
image

在調(diào)用異步方法之后，可以通過循環(huán)判斷異步方法是否執(zhí)行完成。結(jié)果正如我們所預(yù)期，future 所 get 到的是 AsyncResult 返回的字符串。

配置線程池

前面是最簡單的使用方法，使用默認的 TaskExecutor。如果想使用自定義的 Executor，可以結(jié)合 @Configuration 注解的配置方式。


import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
public class TaskPoolConfig {

    @Bean("taskExecutor") // bean 的名稱，默認為首字母小寫的方法名
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10); // 核心線程數(shù)（默認線程數(shù)）
        executor.setMaxPoolSize(20); // 最大線程數(shù)
        executor.setQueueCapacity(200); // 緩沖隊列數(shù)
        executor.setKeepAliveSeconds(60); // 允許線程空閑時間（單位：默認為秒）
        executor.setThreadNamePrefix("taskExecutor-"); // 線程池名前綴
        // 線程池對拒絕任務(wù)的處理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}

線程池的配置很靈活，對核心線程數(shù)、最大線程數(shù)等屬性進行配置。其中，rejection-policy，當(dāng)線程池已經(jīng)達到最大線程數(shù)的時候，如何處理新任務(wù)?？蛇x策略有 CallerBlocksPolicy、CallerRunsPolicy 等。CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是由調(diào)用者所在的線程來執(zhí)行。我們驗證下，線程池的設(shè)置是否生效，在 TaskService 中，打印當(dāng)前的線程名稱：

    public Future<String> doTaskOne() throws Exception {
        log.info("開始做任務(wù)一");
        long start = System.currentTimeMillis();
        Thread.sleep(1000);
        long end = System.currentTimeMillis();
        log.info("完成任務(wù)一，耗時：" + (end - start) + "毫秒");
        log.info("當(dāng)前線程為 {}", Thread.currentThread().getName());
        return new AsyncResult<>("任務(wù)一完成，耗時" + (end - start) + "毫秒");
    }
image

通過結(jié)果可以看到，線程池配置的線程名前綴已經(jīng)生效。在 Spring @Async 異步線程使用過程中，需要注意的是以下的用法會使 @Async 失效：

  • 異步方法使用 static 修飾；
  • 異步類沒有使用 @Component 注解（或其他注解）導(dǎo)致 Spring 無法掃描到異步類；
  • 異步方法不能與被調(diào)用的異步方法在同一個類中；
  • 類中需要使用 @Autowired 或 @Resource 等注解自動注入，不能手動 new 對象；
  • 如果使用 Spring Boot 框架必須在啟動類中增加 @EnableAsync 注解。

線程上下文信息傳遞

很多時候，在微服務(wù)架構(gòu)中的一次請求會涉及多個微服務(wù)。或者一個服務(wù)中會有多個處理方法，這些方法有可能是異步方法。有些線程上下文信息，如請求的路徑，用戶唯一的 userId，這些信息會一直在請求中傳遞。如果不做任何處理，我們看下是否能夠正常獲取這些信息。

@GetMapping("/task")
    public String taskExecute() {
        try {
            Future<String> r1 = taskService.doTaskOne();
            Future<String> r2 = taskService.doTaskTwo();
            Future<String> r3 = taskService.doTaskThree();

            ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
            HttpServletRequest request = requestAttributes.getRequest();
            log.info("當(dāng)前線程為 {}，請求方法為 {}，請求路徑為：{}", Thread.currentThread().getName(), request.getMethod(), request.getRequestURL().toString());
            while (true) {
                if (r1.isDone() && r2.isDone() && r3.isDone()) {
                    log.info("execute all tasks");
                    break;
                }
                Thread.sleep(200);
            }
            log.info("\n" + r1.get() + "\n" + r2.get() + "\n" + r3.get());
        } catch (Exception e) {
            log.error("error executing task for {}", e.getMessage());
        }

        return "ok";
    }

在 Spring Boot Web 中我們可以通過 RequestContextHolder 很方便的獲取 request。在接口方法中，輸出請求的方法和請求的路徑。

    public Future<String> doTaskOne() throws Exception {
        log.info("開始做任務(wù)一");
        long start = System.currentTimeMillis();
        Thread.sleep(1000);
        long end = System.currentTimeMillis();
        log.info("完成任務(wù)一，耗時：" + (end - start) + "毫秒");
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        HttpServletRequest request = requestAttributes.getRequest();
        log.info("當(dāng)前線程為 {}，請求方法為 {}，請求路徑為：{}", Thread.currentThread().getName(), request.getMethod(), request.getRequestURL().toString());
        return new AsyncResult<>("任務(wù)一完成，耗時" + (end - start) + "毫秒");
    }

同時在 TaskService 中，驗證是不是也能輸出請求的信息。運行程序，結(jié)果如下：

image

在 TaskService 中，每個異步線程的方法獲取 RequestContextHolder 中的請求信息時，報了空指針異常。這說明了請求的上下文信息未傳遞到異步方法的線程中。RequestContextHolder 的實現(xiàn)，里面有兩個 ThreadLocal 保存當(dāng)前線程下的 request。

    //得到存儲進去的request
    private static final ThreadLocal<RequestAttributes> requestAttributesHolder =
            new NamedThreadLocal<RequestAttributes>("Request attributes");
    //可被子線程繼承的request
    private static final ThreadLocal<RequestAttributes> inheritableRequestAttributesHolder =
            new NamedInheritableThreadLocal<RequestAttributes>("Request context");

再看 #getRequestAttributes() 方法，相當(dāng)于直接獲取 ThreadLocal 里面的值，這樣就使得每一次獲取到的 Request 是該請求的 request。如何將上下文信息傳遞到異步線程呢？Spring 中的 ThreadPoolTaskExecutor 有一個配置屬性 TaskDecorator，TaskDecorator 是一個回調(diào)接口，采用裝飾器模式。裝飾模式是動態(tài)的給一個對象添加一些額外的功能，就增加功能來說，裝飾模式比生成子類更為靈活。因此 TaskDecorator 主要用于任務(wù)的調(diào)用時設(shè)置一些執(zhí)行上下文，或者為任務(wù)執(zhí)行提供一些監(jiān)視/統(tǒng)計。

public interface TaskDecorator {

    Runnable decorate(Runnable runnable);
}

#decorate 方法，裝飾給定的 Runnable，返回包裝的 Runnable 以供實際執(zhí)行。

下面我們定義一個線程上下文拷貝的 TaskDecorator。

import org.springframework.core.task.TaskDecorator;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;

public class ContextDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        RequestAttributes context = RequestContextHolder.currentRequestAttributes();
        return () -> {
            try {
                RequestContextHolder.setRequestAttributes(context);
                runnable.run();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
}

實現(xiàn)較為簡單，將當(dāng)前線程的 context 裝飾到指定的 Runnable，最后重置當(dāng)前線程上下文。

在線程池的配置中，增加回調(diào)的 TaskDecorator 屬性的配置：

    @Bean("taskExecutor")
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(200);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("taskExecutor-");
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        // 增加 TaskDecorator 屬性的配置
        executor.setTaskDecorator(new ContextDecorator());
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

經(jīng)過如上配置，我們再次運行服務(wù)，并訪問接口，控制臺日志信息如下：

image

由結(jié)果可知，線程的上下文信息傳遞成功。

小結(jié)

本文結(jié)合示例講解了 Spring 中實現(xiàn)異步方法，獲取異步方法的返回值。并介紹了配置 Spring 線程池的方式。最后介紹如何在異步多線程中傳遞線程上下文信息。線程上下文傳遞在分布式環(huán)境中會經(jīng)常用到，比如分布式鏈路追蹤中需要一次請求涉及到的 TraceId、SpanId。簡單來說，需要傳遞的信息能夠在不同線程中。異步方法是我們在日常開發(fā)中用來多線程處理業(yè)務(wù)邏輯，這些業(yè)務(wù)邏輯不需要嚴格的執(zhí)行順序。用好異步解決問題的同時，更要用對異步多線程的方式。

源碼地址

推薦閱讀

微服務(wù)合集

訂閱最新文章，歡迎關(guān)注我的公眾號

微信公眾號
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容