SpringCloud解決Feign異步回調(diào)問題(SpringBoot+Async+Future實(shí)現(xiàn))

更新時(shí)間：2022年11月23日 11:17:03 作者：弱水提滄

這篇文章主要介紹了SpringCloud解決Feign異步回調(diào)問題(SpringBoot+Async+Future實(shí)現(xiàn))，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

一、背景

對于一個(gè)頁面上的所有實(shí)例有一個(gè)查詢權(quán)限的接口，同時(shí)有四個(gè)操作類型的需要查詢且接口僅支持單個(gè)實(shí)例單個(gè)操作類型操作。

簡單來說，假設(shè)實(shí)例查詢退訂權(quán)限需要1秒鐘，那么四個(gè)操作共需4秒鐘，一共20個(gè)實(shí)例的話，那么實(shí)際所需時(shí)間為80秒。

這樣顯然是不符合要求的。經(jīng)過本文的優(yōu)化后，可以達(dá)到20秒。

二、設(shè)計(jì)方案

需要指出，由于各種限制原因，無法直接在工程中進(jìn)行rest接口的調(diào)用，否則是直接可以@Async異步調(diào)用的。

（一）使用框架

產(chǎn)品工程使用SpringCloud微服務(wù)，同時(shí)通過Feign調(diào)用權(quán)限的微服務(wù)，服務(wù)都是使用SpringBoot實(shí)現(xiàn)。

（二）實(shí)現(xiàn)方案

由于Feign的特殊性，目前并不支持異步返回Future，直接通過Future是會(huì)報(bào)錯(cuò)的。

因此可以在權(quán)限微服務(wù)中異步執(zhí)行查詢操作權(quán)限，同時(shí)再異步起一個(gè)線程去等待所有的異步結(jié)果，這一個(gè)線程是阻塞的，同時(shí)由阻塞得到的最終結(jié)果，封裝成自己想要的返回體，返回給調(diào)用方即可，這樣相當(dāng)于大致只需要等待一個(gè)實(shí)例的操作時(shí)間即可。

三、示例代碼

項(xiàng)目代碼我就不放上來了，我將寫一個(gè)Demo作為示例，基本的思想都在里面了。

（一）被調(diào)用方

1. 配置異步化及異步線程池 - AsyncConfig類

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
 
    @Bean("taskPoolExecutor")
    public Executor taskPoolExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(30);
        executor.setQueueCapacity(200);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("taskPoolExecutor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
 
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
}

2. Controller層

@RestController
@Slf4j
public class AsyncController {
 
    @Autowired
    private AsyncService asyncService;
 
    /**
     * 異步測試方法
     * @param ids
     * @return
     * @throws Exception
     */
    @GetMapping("/future")
    public List<User> testFuture(int[] ids) throws Exception {
        List<Future<User>> futures = new ArrayList<>();
        futures.add(asyncService.testFuture(ids[0])); //異步方法 異步執(zhí)行四個(gè)用戶的查詢操作
        futures.add(asyncService.testFuture(ids[1]));
        futures.add(asyncService.testFuture(ids[2]));
        futures.add(asyncService.testFuture(ids[3]));
        return asyncService.getReturnValueMange(futures); 異步方法 異步獲取所有的異步回調(diào)值
    }
 
    /**
     * 同步測試方法
     * @param i
     * @return
     * @throws InterruptedException
     */
    @GetMapping("/syncTest")
    public User testsyncTest( Integer i) throws InterruptedException {
        User user = asyncService.testSync(i);
        return user;
    }
}

這里需要解釋說明下，使用一個(gè)list作為future結(jié)果的集合，同時(shí)采用getReturnValueMange（future）方法來獲取所有的異步執(zhí)行結(jié)果，具體可以看下面Service層的代碼邏輯。

3. AsyncService層

@Slf4j
@Service
public class AsyncService {
 
    @Autowired
    private AsyncUnit asyncUnit; //異步實(shí)現(xiàn)util
 
    public Future<User> testFuture(int i) throws InterruptedException {
        log.info("start...");
        long currentTime = System.currentTimeMillis();
        Future<User> user = asyncUnit.testUser(i);
        log.info("done...{}", String.valueOf(System.currentTimeMillis()- currentTime));
        return user;
    }
 
    //獲取異步方法結(jié)果
    public List<User> getReturnValueMange(List<Future<User>> futures) throws Exception {
        Future<List<User>> userFuture = asyncUnit.getReturnValueMange(futures);
        return userFuture.get(); //get（）方法為阻塞方法，等待異步返回值
    }
 
    //同步方法-作為比較方法
    public User testSync(int i) throws InterruptedException {
        log.info("start...");
        Thread.sleep(2000);
        return new User(i);
    }

4. AsyncUtil 層

@Service
@Slf4j
public class AsyncUnit {
 
    //異步方法，通過sleep 2秒鐘模擬
    @Async("taskPoolExecutor")
    public Future<User> testUser(int age){
        log.info("異步執(zhí)行start...{}",Thread.currentThread().getName());
        long currentTime = System.currentTimeMillis();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("異步執(zhí)行done...{}, {}",Thread.currentThread().getName(), String.valueOf(System.currentTimeMillis()- currentTime));
        return new AsyncResult<>(new User(age));
    }
 
    @Async("taskPoolExecutor")
    public Future<List<User>> getReturnValueMange(List<Future<User>> futures) throws Exception{
        log.info("異步執(zhí)行start..getReturnValueMange.{}",Thread.currentThread().getName());
        long currentTime = System.currentTimeMillis();
        List<User> users = new ArrayList<>();
        for (Future<User> future : futures) {
            users.add(future.get());
        }
        log.info(""+users.size());
        log.info("異步執(zhí)行done..getReturnValueMange.{}, {}",Thread.currentThread().getName(), String.valueOf(System.currentTimeMillis()- currentTime));
        return new AsyncResult<>(users);
    }
}

（二）調(diào)用方

1. Config 層

@Configuration
public class FeignConfig {
 
    @Bean
    public RequestInterceptor requestInterceptor() {
        return new FeignRequestInterceptor();
    }
}

2. Controller層

@RestController
@Slf4j
public class AsyncController {
 
    @Autowired
    private RemoteClient remoteClient;
 
    //異步測試方法，查詢五次 模擬所需時(shí)間
    @GetMapping("/future")
    public List<Integer> testFuture() throws InterruptedException, ExecutionException {
        String res = "";
        log.info(" asynccontroller...start...");
        long currentTime = System.currentTimeMillis();
        int[] ids = new int[]{1,2,3,4,5};
        List<User> users = remoteClient.testFuture(ids);
        List<Integer> resList = new ArrayList<>();
        resList.add(users.get(1).getAge());
        log.info(" asynccontroller...done...{}", String.valueOf(System.currentTimeMillis()- currentTime));
        return  resList;
    }
 
    //同步測試方法，通過執(zhí)行5次同步方法模擬查詢
    @GetMapping("/syncTest")
    public String testsyncTest() {
        long currentTime = System.currentTimeMillis();
        List<Integer> ages = new ArrayList<>();
        for(int i=0; i<5; i++){
            User user = remoteClient.testsyncTest(i);
            ages.add(user.getAge());
        }
        log.info("done...{}", String.valueOf(System.currentTimeMillis()- currentTime));
        return  ages.get(0)+ages.get(1)+ages.get(2)+ages.get(3)+ages.get(4)+"";
    }
}

3. FeignClient （RemoteClient）層

@FeignClient(name = "ASYNC")
public interface RemoteClient {
 
 
    @GetMapping("/future")
    List<User> testFuture(@RequestParam("ids") int[] ids);
 
 
    @GetMapping("/futureValue")
    List<String> getReturnValueMange(@RequestBody List<Future<User>> futures);
 
    @GetMapping("/syncTest")
    User testsyncTest(@RequestParam("i") Integer i);
 
}

被調(diào)用方的服務(wù)名為ASYNC，通過Feign調(diào)用，F(xiàn)eign的詳細(xì)說明就不細(xì)說，具體可以自行查詢。

（三）公共類 User

@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {
 
    private String name;
    private String id;
    private int age;
    private String school;
 
    public User (int age){
        this.age = age;
    }
 
}