Spring?Cloud?Gateway調用Feign異步問題怎么解決

這篇文章主要講解了“Spring Cloud Gateway調用Feign異步問題怎么解決”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Spring Cloud Gateway調用Feign異步問題怎么解決”吧！

版本設定 spring cloud 2020.0.2版本

HttpMessageConverters

原因

由于Spring Cloud Gateway 是基于Spring 5、Spring Boot 2.X和Reactor開發的響應式組件，運用了大量的異步實現。

在項目啟動過程中，并不會創建HttpMessageConverters實例。

解決方法

啟動時創建相應的Bean,注入到Spring容器

@Configuration
public class FeignConfig {

    @Bean
    public Decoder decoder(){
        return new ResponseEntityDecoder(new SpringDecoder(feignHttpMessageConverter()));
    }
    private ObjectFactory<HttpMessageConverters> feignHttpMessageConverter(){
        HttpMessageConverters httpMessageConverters=new HttpMessageConverters
                (new MappingJackson2HttpMessageConverter());
        return ()->httpMessageConverters;
    }
}

Filter異步調用問題

場景

以鑒權為例，外部訪問經由Gateway路由轉發，需要驗證當前請求中是否存在token，可以通過自定義過濾器實現GlobalFitler實現。

@PropertySource(value = "classpath:loginfilter.properties")
@Component
public class AuthLoginGlobalFilter implements GlobalFilter, Ordered {
    @Value("#{'/per-user/login,/goods/**'.split(',')}")
    private List<String> ignoreUrls;
    @Autowired
    private IUserFeign userFeign;
    ExecutorService executorService = Executors.newFixedThreadPool(1);
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        if(ignoreUrls !=null && ignoreUrls.contains(request.getURI().getPath())) {
            return chain.filter(exchange);
        }
        String access_token = request.getHeaders().getFirst("access_token");
        if(StringUtils.isBlank(access_token)) {
            return onError(exchange,"尚未登錄");
        }
        R<String> r = userFeign.validToken(access_token);
        if(r.getCode() == 200) {
            ServerHttpRequest serverHttpRequest = request.mutate().header("uid",r.getData()).build();
            return chain.filter(exchange.mutate().request(serverHttpRequest).build());
        }

        return onError(exchange,r.getMsg());
    }

    @Override
    public int getOrder() {
        return 0;
    }

    private Mono<Void> onError(ServerWebExchange exchange,String msg) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.UNAUTHORIZED);
        response.getHeaders().add("Content-Type","application/json;charset=UTF-8");
        R r = new R.Builder().buildCustomize(HttpStatus.UNAUTHORIZED.value(),msg);
        ObjectMapper objectMapper = new ObjectMapper();
        String rs = "";
        try {
            rs = objectMapper.writeValueAsString(r);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        DataBuffer dataBuffer =response.bufferFactory().wrap(rs.getBytes());
        return response.writeWith(Flux.just(dataBuffer));
    }
}

R r = userFeign.validToken(access_token);屬于同步調用，會報以下錯誤：

錯誤原因

在BlockingSingleSubscriber中會進行判斷：

final T blockingGet() {
		if (Schedulers.isInNonBlockingThread()) {
			throw new IllegalStateException("block()/blockFirst()/blockLast() are blocking, which is not supported in thread " + Thread.currentThread().getName());
		}
		if (getCount() != 0) {
			try {
				await();
			}
			catch (InterruptedException ex) {
				dispose();
				throw Exceptions.propagate(ex);
			}
		}

		Throwable e = error;
		if (e != null) {
			RuntimeException re = Exceptions.propagate(e);
			//this is ok, as re is always a new non-singleton instance
			re.addSuppressed(new Exception("#block terminated with an error"));
			throw re;
		}
		return value;
	}

解決方案

解決方案，同步轉異步，如果需要獲取返回結果，可以通過Future方式獲取

@PropertySource(value = "classpath:loginfilter.properties")
@Component
public class AuthLoginGlobalFilter implements GlobalFilter, Ordered {
    @Value("#{'/per-user/login,/goods/**'.split(',')}")
    private List<String> ignoreUrls;
    @Autowired
    private IUserFeign userFeign;
    ExecutorService executorService = Executors.newFixedThreadPool(1);
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        if(ignoreUrls !=null && ignoreUrls.contains(request.getURI().getPath())) {
            return chain.filter(exchange);
        }
        String access_token = request.getHeaders().getFirst("access_token");
        if(StringUtils.isBlank(access_token)) {
            return onError(exchange,"尚未登錄");
        }
        // WebFlux異步調用，同步會報錯
        Future future = executorService.submit((Callable<R>) () -> userFeign.validToken(access_token));
        R<String> r = null;
        try {
            r = (R<String>) future.get();
            if(r.getCode() == 200) {
                ServerHttpRequest serverHttpRequest = request.mutate().header("uid",r.getData()).build();
                return chain.filter(exchange.mutate().request(serverHttpRequest).build());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return onError(exchange,r.getMsg());
    }

    @Override
    public int getOrder() {
        return 0;
    }

    private Mono<Void> onError(ServerWebExchange exchange,String msg) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.UNAUTHORIZED);
        response.getHeaders().add("Content-Type","application/json;charset=UTF-8");
        R r = new R.Builder().buildCustomize(HttpStatus.UNAUTHORIZED.value(),msg);
        ObjectMapper objectMapper = new ObjectMapper();
        String rs = "";
        try {
            rs = objectMapper.writeValueAsString(r);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        DataBuffer dataBuffer =response.bufferFactory().wrap(rs.getBytes());
        return response.writeWith(Flux.just(dataBuffer));
    }
}

感謝各位的閱讀，以上就是“Spring Cloud Gateway調用Feign異步問題怎么解決”的內容了，經過本文的學習后，相信大家對Spring Cloud Gateway調用Feign異步問題怎么解決這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！