        <dependency>
            <groupId>com.me</groupId>
            <artifactId>lsf-client</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

2、api 包定义

3、接口实现

?4、提供者 spring bean 配置

    <lsf:registry id="registry" host="127.0.0.1" port="25000"/>

    <lsf:provider id="helloWorldServiceLsf" alias="test"
                  interface="com.me.lsf.provider.api.HelloWorldService"
                  registry="registry"
                  ref="helloWorldService"/>

5、启动类

四、调用方 demo

?1、引入客户端 jar 包和提供者的 api 包

        <dependency>
            <groupId>com.me</groupId>
            <artifactId>lsf-provider-api</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>com.me</groupId>
            <artifactId>lsf-client</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

2、消费者 spring bean 配置

    <lsf:registry id="registry" host="127.0.0.1" port="25000"/>

    <lsf:consumer id="helloWorldService"
                  interface="com.me.lsf.provider.api.HelloWorldService"
                  registry="registry"
                  alias="test"/>

3、启动类

?五、具体实现

1、注册中心、消费者、生产证 spring bean标签定义

(1) lsf.xsd 文件

<?xml version="1.0" encoding="UTF-8" ?>
<schema xmlns="http://www.w3.org/2001/XMLSchema"
        targetNamespace="http://www.me.com/schema/lsf"
        elementFormDefault="qualified">

    <element name="registry">
        <complexType>
            <attribute name="id" type="string"/>
            <attribute name="host" type="string"/>
            <attribute name="port" type="string"/>
        </complexType>
    </element>
    <element name="provider">
        <complexType>
            <attribute name="id" type="string"/>
            <attribute name="alias" type="string"/>
            <attribute name="interface" type="string"/>
            <attribute name="ref" type="string"/>
            <attribute name="registry" type="string"/>
        </complexType>
    </element>
    <element name="consumer">
        <complexType>
            <attribute name="id" type="string"/>
            <attribute name="alias" type="string"/>
            <attribute name="interface" type="string"/>
            <attribute name="registry" type="string"/>
        </complexType>
    </element>
</schema>

(2) 标签解析

2、核心数据结构

(1) rpc参数

public class RpcParam {

    /**
     * 调用类
     */
    private String rClass;
    
    /**
     * 调用方法
     */
    private String method;
    
    /**
     * 参数列表
     */
    private String[] args;
    
    /**
     * 序列化方式
     */
    private String serializeType = SerializeTypeEnum.JSON_AUTO_TYPE.getCode();
}

?(2) 消费者bean


public class Consumerbean {

    /**
     * 调用接口
     */
    private Class interfaceClass;

    /**
     * 调用接口名
     */
    private String interfaceName;

    /**
     * 别名
     */
    private String alias;

    /**
     * 预留
     */
    private Boolean register;

    /**
     * 存活生产者连接
     */
    private List<LsfConnection> aliveConnectionList;

    /**
     * 手工指定生产者连接
     */
    private List<LsfConnection> fixedConnectionList;

    /**
     * 父对象
     */
    private ParentObject parentObject;

    /**
     * 序列化方式
     */
    private String serializeType = SerializeTypeEnum.JSON_AUTO_TYPE.getCode();

    /**
     * 注册中心 bean
     */
    private RegistryBean registryBean;
}

3、核心处理逻辑

(1) 消费核心处理逻辑

public class ConsumerBeanInvocationHandler implements InvocationHandler {

    private static Logger logger = LoggerFactory.getLogger(ConsumerBeanInvocationHandler.class);

    private Consumerbean consumerbean;

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        //获取调用的类
        Class tClass = consumerbean.getInterfaceClass();

        String canonicalName = tClass.getCanonicalName();
        String methodName = method.getName();

        ParentObject parentObject = consumerbean.getParentObject();

        boolean isNoRpc = isNoRpc(methodName, parentObject);

        if (isNoRpc) {
            return method.invoke(parentObject, args);
        }
        logger.debug("执行了 rpc 调用, class {}, method {}, args {}",canonicalName, methodName, Arrays.toString(args));

        //获取序列化方式
        String serializeType = consumerbean.getSerializeType();
        
        //组装 rpc 参数
        RpcParam rpcParam = getRpcParam(args, canonicalName, methodName, serializeType, method);

        //获取可用的生产者连接
        LsfConnection connection = consumerbean.getConnection();

        //调用并得到字符串结果
        String rpcResponseParamStr = getBody(rpcParam, connection);

        RpcResponseParam rpcResponseParam = JSON.parseObject(rpcResponseParamStr, RpcResponseParam.class);

        if (ErrorCodeEnum.SUCCESS.getCode().equals(rpcResponseParam.getCode())) {
            //获取序列化处理类
            LsfSerialize lsfSerialize = LsfSerializeFactory.get(serializeType);
            //反序列化结果
            Object result = lsfSerialize.deSerializeResult(method, rpcResponseParam.getResult());
            return result;
        } else {
            //生产者抛出的异常处理
            throw new RuntimeException(rpcResponseParam.getException());
        }

    }

    private String getBody(RpcParam rpcParam, LsfConnection connection) {
        LsfClient client = LsfHttpClientFactory.getClient();
        String host = connection.getHost();
        int port = connection.getPort();
        ClientParam clientParam = new ClientParam();
        clientParam.setHost(host);
        clientParam.setPort(port);
        clientParam.setUrl("/");

        String rpcBody = JSON.toJSONString(rpcParam);
        clientParam.setBody(rpcBody);
        // netty 执行网络调用
        return client.post(clientParam);
    }


    private RpcParam getRpcParam(Object[] args, String canonicalName, String methodName, String serializeType, Method method) {
        RpcParam rpcParam = new RpcParam();

        rpcParam.setrClass(canonicalName);
        rpcParam.setMethod(methodName);
        rpcParam.setSerializeType(serializeType);

        //获取序列化方式
        LsfSerialize lsfSerialize = LsfSerializeFactory.get(serializeType);

        //序列化参数
        String[] argsStrs = lsfSerialize.serializeParam(method, args);

        rpcParam.setArgs(argsStrs);
        return rpcParam;
    }

    private boolean isNoRpc(String methodName, ParentObject parentObject) {
        boolean isNoRpc = false;
        Method[] declaredMethods = parentObject.getClass().getDeclaredMethods();
        for (Method declaredMethod : declaredMethods) {
            if (declaredMethod.getName().equals(methodName)) {
                isNoRpc = true;
                break;
            }
        }
        return isNoRpc;
    }
}

(2) 生产者核心处理逻辑

public static String dealRequest(String body) {

        logger.info("center asyncDeal request {}",body);

        //解析 rpc 调用参数
        RpcParam rpcParam = JSON.parseObject(body, RpcParam.class);

        String rClassStr = rpcParam.getrClass();
        Object provider = getProvider(rClassStr);

        if (provider == null) {
            throw new RuntimeException("没有 对应的 provider");
        }

        String method = rpcParam.getMethod();
        String[] argsStr = rpcParam.getArgs();

        Class<?> aClass = provider.getClass();

        String resultStr = "error";
        RpcResponseParam rpcResponseParam = new RpcResponseParam();
        try {
            Method declaredMethod = null;

            //通过反射获取rpc调用的方法
            Method[] declaredMethods = aClass.getDeclaredMethods();
            for (Method declaredMethod1 : declaredMethods) {
                if (declaredMethod1.getName().equals(method)) {
                    declaredMethod = declaredMethod1;
                    break;
                }
            }

            if (declaredMethod == null) {
                throw new RuntimeException("没有这个方法 " + method);
            }

            //获取序列化实现类
            LsfSerialize lsfSerialize = LsfSerializeFactory.get(rpcParam.getSerializeType());

            //反序列参数
            Object[] inArgs = lsfSerialize.deSerializeParam(declaredMethod, argsStr);

            //调用实现类
            Object result = declaredMethod.invoke(provider, inArgs);
            
            //序列化执行结果
            String result1 = lsfSerialize.serializeResult(declaredMethod, result);
            rpcResponseParam.setCode(ErrorCodeEnum.SUCCESS.getCode());
            rpcResponseParam.setResult(result1);

        } catch (Exception e) {
            //若原始方法发生异常，则封装异常信息并返回给消费者
            rpcResponseParam.setCode(ErrorCodeEnum.EXCEPTION.getCode());
            rpcResponseParam.setException(e.toString());
            logger.error("lsf rpc exception rpc param {}", JSON.toJSONString(rpcParam), e);
        }
        resultStr = JSON.toJSONString(rpcResponseParam);

        logger.info("center asyncDeal result {}", resultStr);
        return resultStr;
    }

(3) 序列化接口定义

对扩展开放。新的的序列化方式可通过实现这个接口，并注册到序列化工厂里去实现。

public interface LsfSerialize {

    String[] serializeParam(Method method, Object[] args);

    Object[] deSerializeParam(Method method, String[] contents);

    String serializeResult(Method method, Object result);

    Object deSerializeResult(Method method, String content);

}

(4) FastJson autoType 序列化实现


public class JsonAutoTypeSerialize implements LsfSerialize {

    private static Logger logger = LoggerFactory.getLogger(JsonAutoTypeSerialize.class);

    {
        ParserConfig.getGlobalInstance().setSafeMode(false);
        ParserConfig.getGlobalInstance().setAutoTypeSupport(true);
    }

    @Override
    public String[] serializeParam(Method method, Object[] args) {

        String[] argsStrs = null;
        if (args != null) {
            argsStrs = new String[args.length];
            for (int i = 0; i < args.length; i++) {
                argsStrs[i] = JSON.toJSONString(args[i], SerializerFeature.WriteClassName);
            }
        }
        return argsStrs;
    }

    @Override
    public Object[] deSerializeParam(Method method, String[] contents) {
        Class<?>[] parameterTypes = method.getParameterTypes();
        return getInArgs(contents, parameterTypes, method);
    }

    private Object[] getInArgs(String[] strs, Class<?>[] parameterTypes, Method method) {

        if (strs == null) {
            return null;
        }
        Type[] genericParameterTypes = method.getGenericParameterTypes();
        Object[] inArgs = new Object[strs.length];
        for (int i = 0; i < parameterTypes.length; i++) {
            Class<?> parameterType = parameterTypes[i];
            inArgs[i] = getObjectSuper(strs[i], parameterType, genericParameterTypes[i]);
        }
        return inArgs;
    }

    @Override
    public String serializeResult(Method method, Object result) {
        return JSON.toJSONString(result, SerializerFeature.WriteClassName);
    }

    @Override
    public Object deSerializeResult(Method method, String content) {
        Type genericReturnType = method.getGenericReturnType();
        Class<?> returnType = method.getReturnType();
        return getObjectSuper(content, returnType, genericReturnType);
    }

    private Object getObjectSuper(String content, Class<?> returnType, Type genericReturnType) {
        Object result = JSON.parseObject(content, returnType);
        return result;
    }
}