06 - WebFlux Client

设计思路

模拟一个类似 Feign 客户端

编写 IUser 接口

@Component
@ApiServer("http://localhost:8080/user")
public interface IUserApi {
    @GetMapping("/all")
    Flux<User> getAll();
    @GetMapping("/get/{id}")
    Mono<User> getById(@PathVariable("id") String id);
    @DeleteMapping("/delete/{id}")
    Mono<Void> deleteById(@PathVariable("id") String id);
    @PostMapping("/create")
    Mono<User> create(@RequestBody Mono<User> user);
}

这里我们使用了一个自定义注解 @ApiServer 用来保存 URL 前缀

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface ApiServer {
    String value() default "";
}

对应的，还使用了一个 Domain User，用来接收和输入

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class User {
    private String id;
    private String name;
    private int age;
}

编写 Proxy 接口

调用接口 IUserApi 我们编写完成后，我们就需要对 IUserApi 进行代理，获取方法的参数，返回值等，并进行调用，由于代理可能会使用 JDK 动态代理，也可能使用 CGLIB 代理，所以我们先编写 Proxy 接口，接口很简单，只有一个方法，就是创建代理对象

public interface ProxyCreator {
    Object create(Class<?> clazz);
}

需要使用到的两个 POJO

ServerInfo

ServerInfo，用来记录服务器信息，目前只记录服务器前缀，配合 @ApiServer 使用

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MethodInfo {
    /** 请求 URL */
    private String url;
    /** 请求方法 */
    private HttpMethod method;
    /** 请求参数 */
    private Map<String, Object> params;
    /** 请求体 */
    private Mono<?> body;
    /** 请求体的类型 */
    private Class<?> bodyElementType;
    /** 判断是 Flux 还是 Mono */
    private boolean returnFlux;
    /** 返回对象的类型 */
    private Class<?> returnElementType;
}

MethodInfo

MethodInfo，用来描述方法信息，正对 IUserApi 接口中的方法

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MethodInfo {
    /** 请求 URL */
    private String url;
    /** 请求方法 */
    private HttpMethod method;
    /** 请求参数 */
    private Map<String, Object> params;
    /** 请求体 */
    private Mono<?> body;
    /** 请求体的类型 */
    private Class<?> bodyElementType;
    /** 判断是 Flux 还是 Mono */
    private boolean returnFlux;
    /** 返回对象的类型 */
    private Class<?> returnElementType;
}

编写 RestHandler 请求接口

Rest 请求可以由 WebClient 来实现，也可以有 Spring 的 RestTemplate 来实现，所以为了更好的扩展，我们也抽象成接口的形式

public interface RestHandler {
    /**
     * 初始化服务器信息
     *
     * @param serverInfo 服务器信息
     */
    void init(ServerInfo serverInfo);
    /**
     * 调用 REST 请求
     *
     * @param methodInfo 方法信息
     * @return Object
     */
    Object invoke(MethodInfo methodInfo);
}

编写 RestHandler 实现类（WebClientRestHandler）

这里我们使用 WebClient 来实现 RestHandler 接口

public class WebClientRestHandler implements RestHandler {
    private WebClient webClient;
    @Override
    public void init(ServerInfo serverInfo) {
        webClient = WebClient.create(serverInfo.getUrl());
    }
    /**
     * 处理 Rest 请求
     *
     * @param methodInfo 方法信息
     */
    @Override
    public Object invoke(MethodInfo methodInfo) {
        Object result;
        WebClient.RequestBodySpec bodySpec = this.webClient
                // 请求方法
                .method(methodInfo.getMethod())
                // 请求 URI
                .uri(methodInfo.getUrl(), methodInfo.getParams())
                // 响应格式
                .accept(MediaType.APPLICATION_JSON);
        // 判断是否含有 body
        if (methodInfo.getBody() != null) {
            bodySpec.body(methodInfo.getBody(), methodInfo.getBodyElementType());
        }
        // 发送请求
        WebClient.ResponseSpec request = bodySpec.retrieve();
        // 异常处理
        request.onStatus(httpStatus -> httpStatus.value() == 404, response -> Mono.just(new RuntimeException("Not Found")));
        // 处理响应
        if (methodInfo.isReturnFlux()) {
            result = request.bodyToFlux(methodInfo.getReturnElementType());
        } else {
            result = request.bodyToMono(methodInfo.getReturnElementType());
        }
        return result;
    }
}

编写 Proxy 实现类（JDKProxyCreator）

这里我们使用 JDK 动态代理来实现

@Slf4j
public class JDKProxyCreator implements ProxyCreator {
    @Override
    public Object create(Class<?> clazz) {
        log.info("createProxy: {}", clazz);
        // 得到接口的 API 服务器地址
        ServerInfo serverInfo = extractServerInfo(clazz);
        log.info("serverInfo: {}", serverInfo);
        // 给每一个代理类一个实现
        RestHandler restHandler = new WebClientRestHandler();
        // 初始化服务器信息和 WebClient
        restHandler.init(serverInfo);
        return Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class<?>[]{clazz}, (proxy, method, args) -> {
            // 根据方法和参数得到调用信息
            MethodInfo methodInfo = extractMethodInfo(method, args);
            log.info("methodInfo: {}", methodInfo);
            // 调用 REST
            return restHandler.invoke(methodInfo);
        });
    }
    private ServerInfo extractServerInfo(Class<?> clazz) {
        ApiServer apiServer = clazz.getAnnotation(ApiServer.class);
        ServerInfo serverInfo = new ServerInfo();
        serverInfo.setUrl(apiServer.value());
        return serverInfo;
    }
    private MethodInfo extractMethodInfo(Method method, Object[] args) {
        MethodInfo methodInfo = new MethodInfo();
        Annotation[] annotations = method.getAnnotations();
        // 1、获取请求方法和请求URL
        extractRequestUrl(annotations, methodInfo);
        // 2、获取请求参数
        extractRequestParam(method, args, methodInfo);
        // 3、获取返回对象信息
        extractResponseInfo(method, methodInfo);
        return methodInfo;
    }
    /**
     * 获取返回对象信息
     *
     * @param method     method
     * @param methodInfo methodInfo
     */
    private void extractResponseInfo(Method method, MethodInfo methodInfo) {
        // 返回 flux 还是 mono
        // isAssignableFrom 判断类型是否是某个类的子类
        // instanceof 判断实例是否是某个类的子类
        // getReturnType 不带泛型信息
        boolean isFlux = method.getReturnType().isAssignableFrom(Flux.class);
        methodInfo.setReturnFlux(isFlux);
        // getGenericReturnType 带泛型信息
        Type actualType = getActualType(method.getGenericReturnType());
        methodInfo.setReturnElementType((Class<?>) actualType);
    }
    /**
     * 获取请求方法和请求URL
     *
     * @param method     method
     * @param args       参数
     * @param methodInfo methodInfo
     */
    private void extractRequestParam(Method method, Object[] args, MethodInfo methodInfo) {
        Map<String, Object> params = new LinkedHashMap<>();
        methodInfo.setParams(params);
        Parameter[] parameters = method.getParameters();
        for (int i = 0; i < parameters.length; i++) {
            PathVariable pathVariable = parameters[i].getAnnotation(PathVariable.class);
            if (pathVariable != null) {
                params.put(pathVariable.value(), args[i]);
            }
            RequestBody requestBody = parameters[i].getAnnotation(RequestBody.class);
            if (requestBody != null) {
                methodInfo.setBody((Mono<?>) args[i]);
                Type actualTypeArgument = getActualType(parameters[i].getParameterizedType());
                methodInfo.setBodyElementType((Class<?>) actualTypeArgument);
            }
        }
    }
    /**
     * 获取参数实际类型
     *
     * @param type 类型
     */
    private Type getActualType(Type type) {
        ParameterizedType parameterizedType = (ParameterizedType) type;
        return parameterizedType.getActualTypeArguments()[0];
    }
    /**
     * 获取请求参数
     *
     * @param annotations annotations
     * @param methodInfo  methodInfo
     */
    private void extractRequestUrl(Annotation[] annotations, MethodInfo methodInfo) {
        for (Annotation annotation : annotations) {
            // GET
            if (annotation instanceof GetMapping) {
                GetMapping mapping = (GetMapping) annotation;
                methodInfo.setUrl(mapping.value()[0]);
                methodInfo.setMethod(HttpMethod.GET);
            }
            // POST
            else if (annotation instanceof PostMapping) {
                PostMapping mapping = (PostMapping) annotation;
                methodInfo.setUrl(mapping.value()[0]);
                methodInfo.setMethod(HttpMethod.POST);
            }
            // PUT
            else if (annotation instanceof PutMapping) {
                PutMapping mapping = (PutMapping) annotation;
                methodInfo.setUrl(mapping.value()[0]);
                methodInfo.setMethod(HttpMethod.PUT);
            }
            // DELETE
            else if (annotation instanceof DeleteMapping) {
                DeleteMapping mapping = (DeleteMapping) annotation;
                methodInfo.setUrl(mapping.value()[0]);
                methodInfo.setMethod(HttpMethod.DELETE);
            }
        }
    }
}

至此，我们就完成了 IUserApi 的动态代理

让 Spring 容器管理 IUserApi 和 ProxyCreator

我们使用 FactoryBean 来产生 IUserApi 对应的代理 bean

@Bean
ProxyCreator jdkProxyCreator() {
    return new JDKProxyCreator();
}
@Bean
FactoryBean<IUserApi> userApi(ProxyCreator proxyCreator) {
    return new FactoryBean<IUserApi>() {
        /**
         * 返回代理对象
         */
        @Override
        public IUserApi getObject() throws Exception {
            return (IUserApi) proxyCreator.create(getObjectType());
        }
        /**
         * 返回接口类型
         */
        @Override
        public Class<?> getObjectType() {
            return IUserApi.class;
        }
    };
}

测试

现在所需要的 ProxyCreator 和 RestHandler 都准备好了，我们就可以开始测试了

@RestController
public class TestController {
    @Autowired
    private IUserApi userApi;
    @GetMapping("/")
    public Flux<User> getAll() {
        // 实现调用 rest 接口的效果
        return userApi.getAll();
    }
    @GetMapping("/get/{id}")
    public Mono<User> getById(@PathVariable("id") String id) {
        Mono<User> mono = userApi.getById(id);
        mono.subscribe(System.out::println, e -> System.err.println(e.getMessage()));
        return mono;
    }
    @DeleteMapping("/delete/{id}")
    public Mono<Void> deleteById(@PathVariable("id") String id) {
        return userApi.deleteById(id);
    }
    @PostMapping("/create")
    public Mono<User> create(@RequestBody User user) {
        return userApi.create(Mono.just(user));
    }
}