手写一个RPC框架实现远程调用

Love The Way You Lie 2022-11-20 05:59 267阅读 0赞

# 手写一个RPC框架实现远程调用 #

一个RPC框架需要解决哪些问题？

*  通信问题
 *  代理问题
 *  序列化问题
 *  服务实例化

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

## 注册中心 ##

注册中心实现服务提供者注册服务，消费者订阅服务。

package com.morris.service.regcenter;
    
    import com.morris.service.vo.RegisterVo;
    import com.morris.service.vo.ServiceAddress;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.io.ObjectInputStream;
    import java.io.ObjectOutputStream;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.HashSet;
    import java.util.Map;
    import java.util.Set;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    @Slf4j
    public class RegCenterApplication { 
    
        private static final int PORT = 9090;
    
        private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    
        private static Map<String, Set<ServiceAddress>> serviceMap = new ConcurrentHashMap<>(); // cached service
    
        public static void main(String[] args) throws IOException { 
    
            ServerSocket serverSocket = new ServerSocket(PORT);
    
            while (true) { 
                Socket socket = serverSocket.accept();
                executorService.submit(() -> { 
                    try (
                            ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
                            ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream())
                    ) { 
                        final boolean isRegister = ois.readBoolean();
    
                        if(isRegister) { 
                            // register service
                            final RegisterVo registerVo = (RegisterVo) ois.readObject();
    
                            for (String s : registerVo.getServiceList()) { 
                                final ServiceAddress serviceAddress = new ServiceAddress(registerVo.getHost(), registerVo.getPort());
                                if(serviceMap.containsKey(s)) { 
                                    serviceMap.get(s).add(serviceAddress);
                                } else { 
                                    final HashSet<ServiceAddress> hashSet = new HashSet<>();
                                    hashSet.add(serviceAddress);
                                    serviceMap.put(s, hashSet);
                                }
                                log.info("register {} at {}", s, serviceAddress);
                            }
                            oos.writeUTF("success");
                            oos.flush();
                        } else { 
                            // subscribe service
                            final String serviceName = ois.readUTF(); // service name
    
                            final Set<ServiceAddress> serviceAddresses = serviceMap.get(serviceName);
    
                            if(null == serviceAddresses || serviceAddresses.isEmpty()) { 
                                throw new ClassNotFoundException(serviceName + " is not found.");
                            }
    
                            oos.writeObject(serviceAddresses);
                            oos.flush();
                        }
                    } catch (IOException | ClassNotFoundException e) { 
                        e.printStackTrace();
                    }
                });
            }
        }
    
    }

## 服务提供者注册服务 ##

final Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(Provider.class);
    try (
            final Socket socket = new Socket("127.0.0.1", 9090);
            final ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
            final ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
    ) { 
    
        final ArrayList<String> serviceList = new ArrayList<>();
    
        RegisterVo registerVo = new RegisterVo();
        registerVo.setPort(7777);
        registerVo.setHost("127.0.0.1");
        registerVo.setServiceList(serviceList);
    
        for (Object value : beansWithAnnotation.values()) { 
            for (Class<?> anInterface : value.getClass().getInterfaces()) { 
                serviceList.add(anInterface.getName());
            }
        }
    
        objectOutputStream.writeBoolean(true);
        objectOutputStream.writeObject(registerVo);
        objectOutputStream.flush();
    
        final String s = objectInputStream.readUTF();
        log.info("register result: {}", s);
    } catch (UnknownHostException e) { 
        e.printStackTrace();
    } catch (IOException e) { 
        e.printStackTrace();
    }

## 服务提供者提供服务 ##

final ServerSocket serverSocket = new ServerSocket(7777);
    
    while (true) { 
        final Socket socket = serverSocket.accept();
    
        new Thread(() -> { 
            try (ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
                 ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream())) { 
    
                final String serviceName = ois.readUTF();
                final String methodName = ois.readUTF();
                final Class[] parameterTypes = (Class[]) ois.readObject();
                final Object[] parameters = (Object[]) ois.readObject();
    
                final Class<?> aClass = Class.forName(serviceName);
                final Object bean = applicationContext.getBean(aClass);
    
                final Method method = aClass.getMethod(methodName, parameterTypes);
    
                final Object result = method.invoke(bean, parameters);
                oos.writeObject(result);
                oos.flush();
            } catch (IOException | ClassNotFoundException | NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { 
                e.printStackTrace();
            }
        }).start();
    
    }

## 消费者订阅并消费服务 ##

@Bean
    public PayService payService() throws IOException { 
    
        try (
                final Socket socket = new Socket("127.0.0.1", 9090);
                final ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                final ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream())
        ) { 
            objectOutputStream.writeBoolean(false); // consumer
            objectOutputStream.writeUTF(PayService.class.getName()); // service name
            objectOutputStream.flush();
            final Set<ServiceAddress> serviceAddresses = (Set<ServiceAddress>) objectInputStream.readObject();
    
            log.info("get service list {}", serviceAddresses);
    
            return (PayService) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[]{ PayService.class}, (proxy, method, args) -> { 
    
                final List<ServiceAddress> collect = serviceAddresses.stream().collect(Collectors.toList());
    
                final ServiceAddress serviceAddress = collect.get(new Random().nextInt(collect.size()));
                try (
                        final Socket s = new Socket(serviceAddress.getHost(), serviceAddress.getPort());
                        final ObjectOutputStream oos = new ObjectOutputStream(s.getOutputStream());
                        final ObjectInputStream ois = new ObjectInputStream(s.getInputStream())
                ) { 
                    oos.writeUTF(PayService.class.getName()); // service name
                    oos.writeUTF(method.getName()); // method name
                    oos.writeObject(method.getParameterTypes()); // parameter type
                    oos.writeObject(args); // parameter value
                    oos.flush();
    
                    final Object result = ois.readObject();
                    return result;
                }
            });
    
        } catch (ClassNotFoundException e) { 
            e.printStackTrace();
        }
        return null;
    }

完整代码参考：[rpc-demo][]。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/a170a9198d4d4ede991faab36120b325.png
[rpc-demo]: https://gitee.com/morris131/morris-book/tree/master/back/java/netty/_v_projects/rpc-demo