快捷導(dǎo)航

如何解決異步任務(wù)上下文丟失問題

更新時(shí)間：2024年09月18日 10:48:18 作者：勿語(yǔ)&

在多線程編程中,異步任務(wù)可能會(huì)導(dǎo)致上下文信息丟失,為了解決這個(gè)問題,可以在執(zhí)行異步任務(wù)前,通過自定義TaskDecorator拷貝主線程的上下文至子線程,這樣可以確保上下文在異步執(zhí)行過程中得以保留,將定制的TaskDecorator設(shè)置至線程池,可以有效地解決上下文丟失問題

解決異步任務(wù)上下文丟失問題

上下文丟失主要是因?yàn)橹骶€程和子線程的上下文不能共享。
可以通過執(zhí)行異步任務(wù)之前，將主線程上的上下文信息拷貝到子線程上。

自定義TaskDecorator 來(lái)拷貝主線程上的上下文信息到子線程，然后將自定義的 TaskDecorator實(shí)現(xiàn)類設(shè)置到線程池上。

@Configuration
public class ThreadPoolConfig {

    @Bean(name = "customizeTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor(){
        ThreadPoolTaskExecutor poolExecutor = new ThreadPoolTaskExecutor();
        // 核心線程數(shù)=cpu核心數(shù)+1
        poolExecutor.setCorePoolSize(5);
        // 最大線程數(shù)=cpu核心數(shù)*2
        poolExecutor.setMaxPoolSize(8);
        // 設(shè)置任務(wù)裝飾器
        poolExecutor.setTaskDecorator(taskDecorator());
        // 任務(wù)被拒絕后，交給調(diào)用線程執(zhí)行
        poolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return poolExecutor;
    }

    @Bean
    public TaskDecorator taskDecorator(){
        return new TaskDecorator() {
            @Override
            public Runnable decorate(Runnable runnable) {
                RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
                return ()->{
                    try {
                        RequestContextHolder.setRequestAttributes(requestAttributes);
                        runnable.run();
                    }finally {
                        RequestContextHolder.resetRequestAttributes();
                    }
                };
            }
        };
    }
}

Fegin異步情況丟失上下文問題

在微服務(wù)的開發(fā)中，我們經(jīng)常需要服務(wù)之間的調(diào)用，并且為了提高效率使用異步的方式進(jìn)行服務(wù)之間的調(diào)用，在這種異步的調(diào)用情況下會(huì)有一個(gè)嚴(yán)重的問題，丟失上文下

通過以上圖片可以看出異步丟失上下文的原因是不在同一個(gè)線程，所有數(shù)據(jù)不能共享，Wie了解決這個(gè)問題，我們就需要把之前線程的請(qǐng)求頭上下文，在次存放到其他線程的請(qǐng)求頭上下文就行，具體實(shí)現(xiàn)如下：

案例：feign異步獲取訂單明細(xì)的案例代碼

/**
     * 獲取訂單明細(xì)的vo
     * @return
     */
    @Override
    public OrderConfirmVo orderConfirm() {
        MemberResponseVo member = OrderInterceptor.threadLocal.get();
 
        OrderConfirmVo orderConfirmVo = new OrderConfirmVo();
        System.out.println("主線程："+ Thread.currentThread().getId());
        //獲取主線程的請(qǐng)求頭信息
        RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
 
        //考慮到效率問題  使用異步編排
        CompletableFuture<Void> getAddress = CompletableFuture.runAsync(() -> {
            //子線程中設(shè)置添加主線程的請(qǐng)求頭信息  信息共享  否則遠(yuǎn)程調(diào)用異步處理丟失請(qǐng)求頭信息
            RequestContextHolder.setRequestAttributes(requestAttributes);
            System.out.println("address："+ Thread.currentThread().getId());
            //遠(yuǎn)程獲取地址信息
            List<MemberAddressVo> address = memberFeignService.getAddress(member.getId());
            orderConfirmVo.setAddress(address);
        }, executor);
 
 
        CompletableFuture<Void> getItem = CompletableFuture.runAsync(() -> {
            //子線程中設(shè)置添加主線程的請(qǐng)求頭信息  信息共享  否則遠(yuǎn)程調(diào)用異步處理丟失請(qǐng)求頭信息
            RequestContextHolder.setRequestAttributes(requestAttributes);
            System.out.println("item："+ Thread.currentThread().getId());
            //遠(yuǎn)程獲取購(gòu)物項(xiàng)
            List<OrderItemVo> currentUserCartItems = cartFeignService.getCurrentUserCartItems();
            orderConfirmVo.setItems(currentUserCartItems);
        }, executor).thenRunAsync(()->{
            List<OrderItemVo> items = orderConfirmVo.getItems();
            //獲取所有商品的id
            List<String> collect = items.stream().map(item -> item.getSkuId()).collect(Collectors.toList());
            List<Long> skuIds = collect.stream().map(item -> {
                return Long.parseLong(item);
            }).collect(Collectors.toList());
            R<List<SkuHasStockVo>> skusHasStock = wmsFeignService.getSkusHasStock(skuIds);
            List<SkuHasStockVo> data = skusHasStock.getData(new TypeReference<List<SkuHasStockVo>>() {
            });
            if(data!= null){
                Map<Long, Boolean> collect1 = data.stream().collect(Collectors.toMap(SkuHasStockVo::getSkuId, SkuHasStockVo::getHasStock));
                orderConfirmVo.setStocks(collect1);
            }
        },executor);
 
        //異步編排完成之后執(zhí)行后續(xù)操作
        try {
            CompletableFuture.allOf(getAddress,getItem).get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
 
 
        orderConfirmVo.setIntegration(member.getIntegration());
 
        orderConfirmVo.setPayPrice(orderConfirmVo.getPayPrice());
        orderConfirmVo.setTotal(orderConfirmVo.getTotal());
        //TODO 放重處理  生成token令牌儲(chǔ)存在redis
        String token = UUID.randomUUID().toString().replace("-", "");
        orderConfirmVo.setOrderToken(token);
        redisTemplate.opsForValue().set(OrderConstant.ORDER_TOKEN+member.getId(),token);
 
        return orderConfirmVo;
    }