SpringCloudGateway调用Feign异步问题记录

2023-05-17 20:05:38 调用记录 SpringCloudGateway

HttpMessageConverters

原因

由于spring Cloud Gateway 是基于Spring 5、Spring Boot 2.X和Reactor开发的响应式组件，运用了大量的异步实现。

在项目启动过程中，并不会创建HttpMessageConverters实例，具体可查看源码HttpMessageConvertersAutoConfiguration

解决方法

启动时创建相应的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));
    }
}

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持。

相关文章