Spring Cloud Gateway調(diào)用Feign異步問題記錄

更新時(shí)間：2023年04月26日 10:36:15 作者：@Kong

這篇文章主要介紹了Spring Cloud Gateway調(diào)用Feign異步問題記錄，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

HttpMessageConverters

原因

由于Spring Cloud Gateway 是基于Spring 5、Spring Boot 2.X和Reactor開發(fā)的響應(yīng)式組件，運(yùn)用了大量的異步實(shí)現(xiàn)。

在項(xiàng)目啟動過程中，并不會創(chuàng)建HttpMessageConverters實(shí)例，具體可查看源碼HttpMessageConvertersAutoConfiguration

解決方法

啟動時(shí)創(chuàng)建相應(yīng)的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異步調(diào)用問題

場景

以鑒權(quán)為例，外部訪問經(jīng)由Gateway路由轉(zhuǎn)發(fā)，需要驗(yàn)證當(dāng)前請求中是否存在token，可以通過自定義過濾器實(shí)現(xiàn)GlobalFitler實(shí)現(xiàn)。

@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);屬于同步調(diào)用，會報(bào)以下錯誤：

錯誤原因

在BlockingSingleSubscriber中會進(jìn)行判斷：

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;
	}

解決方案

解決方案，同步轉(zhuǎn)異步，如果需要獲取返回結(jié)果，可以通過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異步調(diào)用，同步會報(bào)錯
        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));
    }
}