一个简单的rpc框架实现（一）

1
2
3
4
public interface PrintText {
    String print(String text);
}

1
2
3
4
5
6
7
public class SystemPrint implements PrintText {
    public String print(String text) {
        System.out.println(text);
        return "系统已打印:" + text;
    }
}

1
2
3
4
5
6
7
8
9
public class SpringClient {

    public static void main(String[] args){
        BeanFactory apx = new ClassPathXmlApplicationContext("classpath:spring-config.xml");
        PrintText pt = (PrintText) apx.getBean("printText");
        System.out.println(pt.print("springClient"));
    }
}

1
2
3
4
5
   协议设计:
   version:版本号|##|cypher:加密串|##| interfaceName:接口bean名称|##| interface:接口全路径|##|method:调用方法|##|params:参数1的类+参数1对象,单数2类+参数2对象,

   Demo:   `version:1.0.0|##|cypher:default|##|interfaceName:proxyPrintText|##|interface:com.tmall.beans.PrintText|##|method:print|##|params:java.lang.String+演示远程调用`

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
public class SocketConsumerProxy implements InvocationHandler {
    private Object target;

    private RemoteDataSource dataSource;

    public SocketConsumerProxy(RemoteDataSource dataSource){
        this.dataSource = dataSource;
    }

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        StringBuffer stream = this.buildRpcRequest(proxy, method, args);

        Object object = null;
        try{
            long startTime =  System.currentTimeMillis();
            System.out.println("New rpc client send " + stream.toString() + "  time:" + startTime);
            // socket connect
            Socket socket=new Socket(dataSource.getIp(), dataSource.getPort());

            // request
            PrintWriter os = new PrintWriter(socket.getOutputStream());
            os.println(stream.toString());
            os.flush();

            // read response
            BufferedReader br = new BufferedReader(
                    new InputStreamReader(socket.getInputStream()));
            object = br.readLine();

            long endTime =  System.currentTimeMillis();
            System.out.println("client read from service:" + object + "  time:" + (endTime - startTime));
        }catch (IOException e){
            e.printStackTrace();
            object = e.toString();
        }

        return object;
    }

    private StringBuffer buildRpcRequest(Object proxy, Method method, Object[] args) {
        StringBuffer buffer = new StringBuffer();
        buffer.append(String.format("version:%s|##|cypher:%s|##|interfaceName:%s|##|interface:%s|##|method:%s|##|params:"
          , dataSource.getVersion(), dataSource.getCypher(), dataSource.getInterfaceName(), dataSource.getInterfaces(), method.getName()));

        for (Object obj : args){
            buffer.append(obj.getClass().getName() + "+" + obj.toString() + ",");
        }
        return buffer;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
public class SocketProvider{
    private RemoteDataSource dataSource;

    public SocketProvider(RemoteDataSource dataSource) {
        this.dataSource = dataSource;
    }

    public Object provide() throws Throwable {
        ServerSocket serverSocket = new ServerSocket(dataSource.getPort());
        while (true) {
            Socket socket = null;
            try {
                //接收客户连接,只要客户进行了连接,就会触发accept();从而建立连接
                socket = serverSocket.accept();
                this.getRpcRequest(socket);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void getRpcRequest(Socket socket) {
        try {
            System.out.println("rpc client accepted " + socket.getInetAddress() + ":" + socket.getPort() + "   time:" + System.currentTimeMillis());
            // 接收服务器的反馈
            BufferedReader br = new BufferedReader(
                    new InputStreamReader(socket.getInputStream()));

            String msg = br.readLine();

            System.out.println("读到远程调用请求:" + msg);
            Object obj  = this.parseRpcClientRequest(msg);

            // 接收服务器的反馈
            PrintWriter os = new PrintWriter(socket.getOutputStream());
            os.println(obj.toString());
            os.flush();

            System.out.println("rpc server return " + obj + ":" + socket.getPort() + "   time:" + System.currentTimeMillis());
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public Object parseRpcClientRequest(String msg) {
        Object result = "";
        if (StringUtils.isEmpty(msg)){
            return result;
        }

        String[] infos = msg.split("\|##\|");
        Map<String, String> infoMap = new HashMap<String, String>();
        for(String info : infos){
            String[] pair = info.split(":");
            infoMap.put(pair[0], pair[1]);
        }
        if (infoMap.isEmpty()){
            return "无调用参数";
        }

        if (!dataSource.getCypher().equals(infoMap.get("cypher"))){
            return "加密串不对";
        }

        if (!dataSource.getVersion().equals(infoMap.get("version"))){
            return "服务版本号不对";
        }

        String interfaces = infoMap.get("interface");
        String interfaceName = infoMap.get("interfaceName");
        String methodName = infoMap.get("method");
        String params = infoMap.get("params");

        if(StringUtils.isEmpty(interfaces) || StringUtils.isEmpty(methodName) || StringUtils.isEmpty(interfaceName)){
            return "无调用接口或者方法 或者接口beanName";
        }

        // bean
        Object obj = RemoteDataSource.beanFactory.getBean(interfaceName);
        if(obj == null){
            return "未找到对应的服务";
        }

        // bean 和 interface对应关系
        boolean isInterfaceRight = false;
        Class<?>[] clazzArray = obj.getClass().getInterfaces();
        for (Class<?> clazz : clazzArray){
            if (clazz.getName().equals(interfaces)){
                isInterfaceRight = true;
                break;
            }
        }
        if (isInterfaceRight == false){
            return "错误的bean Name 和 interface对应关系";
        }


        // 参数对应的类和对象
        String[] paramsArray = params.split(",");
        Class<?>[] paramsClazzArray = new Class<?>[paramsArray.length];
        Object[] paramObjArray = new Object[paramsArray.length];
        try{
            for (int i = 0; i < paramsArray.length; i ++){
                String paramInfo = paramsArray[i];
                if (StringUtils.isEmpty(paramInfo)) {  // 过滤掉多余的逗号
                    continue;
                }

                String[] paramsInfos = paramInfo.split("\+");

                Class c = Class.forName(paramsInfos[0]);
                paramsClazzArray[i] = c;
                paramObjArray[i] = paramsInfos[1];
            }
        }catch (ClassNotFoundException e){
            e.printStackTrace();
            return "未找到参数对应的类名:" +e.toString();
        }


        try {
            Method method = obj.getClass().getMethod(methodName, paramsClazzArray);
            result = method.invoke(obj, paramObjArray);
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
            return "未找到参数对应的方法名称:" +e.toString();
        }
        catch (IllegalAccessException e) {
            e.printStackTrace();
            return "无效的调用:" +e.toString();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
            return "无效的目标地址:" +e.toString();
        }

        result = "我是服务器代理aop, result=" + result;
        return result;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public class CaishengRemoteSimpleConsumerBean implements FactoryBean, InitializingBean{
    private RemoteDataSource remoteDataSource = new RemoteDataSource();  // 远程数据源


    public void afterPropertiesSet() throws Exception {
        remoteDataSource.init();
    }

    public Object getObject() throws Exception {
        return remoteDataSource.getClientObject();
    }

    public Class<?> getObjectType() {
        return remoteDataSource.getClientObject().getClass();
    }

   ……//一些不重要的get,set操作
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class RemoteDataSource {
    private String ip = "127.0.0.1";
    private int port =  9090;              // 端口
    private String interfaces;             // 接口全路径
    private String interfaceName;          // 接口名
    private String version;
    private String cypher ="default";       // 密码
    private Class<?> interfaceClass;       // 类文件

    private SocketRemotionFactory remotionFactory;

    public void init(){
        remotionFactory = new SocketRemotionFactory(this);

        try {
            interfaceClass = Class.forName(interfaces);
        } catch (ClassNotFoundException e) {

        }
    }

    public Object getClientObject(){
        return remotionFactory.getRemoteClientProxy();
    }

     ……//一些不重要的get,set操作
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public class SocketRemotionFactory {
    private RemoteDataSource dataSource;
    public SocketRemotionFactory(RemoteDataSource dataSource) {
        this.dataSource = dataSource;
    }

    public Object getRemoteClientProxy(){

        Object result = null;

        Class<?> clazz = dataSource.getInterfaceClass();
        if (clazz == null){
            return "错误的client 代理，无对应的class";
        }

        Class<?>[] clazzArray = new Class[1];
        clazzArray[0] = clazz;
        try{
            result = Proxy.newProxyInstance(clazz.getClassLoader(), clazzArray, new SocketConsumerProxy(dataSource));
        }catch (Exception e){
            e.printStackTrace();
        }

        return result;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class CaishengRemoteSimpleProviderBean implements FactoryBean, InitializingBean{
    private RemoteDataSource remoteDataSource = new RemoteDataSource();  // 远程数据源

    private SocketProvider provider;

    public void init() throws Exception {
        this.afterPropertiesSet();
    }

    public void afterPropertiesSet() throws Exception {
        remoteDataSource.init();

        provider = new SocketProvider(remoteDataSource);
    }

    public Object getObject() throws Exception {
        try {
            provider.provide();
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
        return null;
    }

   ……//一些不重要的get,set操作
}

1
2
3
4
5
6
7
8
public class Client {
    public static void main(String[] args){
        BeanFactory apx = new ClassPathXmlApplicationContext("classpath:spring-config.xml");
        PrintText pt = (PrintText) apx.getBean("caishengSimpleRemoteProxyPrintText");
        System.out.println(pt.print("远程调用演示"));
    }
}