阅读 2216

你真的明白RPC 吗?一起来探究 RPC 的实质

​ 不论你是科班出身还是半路转行,这么优秀的你一定上过小学语文,那么对扩句和缩句你一定不陌生。缩句就是去除各种修饰提炼出一句话的核心,而不失基本的语义。下面来实现一个简易的 rpc 程序探究其实质,进而去理解复杂的 rpc 框架。所谓复杂的框架就是在简单的过程中加入了一些设计装饰将rpc的功能丰富起来,如 dubbo 的 filter、router、loadblance、集群容错、多种 Invoker 、通讯协议等等,这就是一个扩句的过程。 文中福利,附一张刘秘美照

RPC是指远程过程调用,也就是说两台服务器A、B,一个应用部署在A服务器上,想要调用B服务器上应用提供的函数/方法,由于不在一个内存空间,不能直接调用,需要通过网络去发起一次调用请求获取结果。

​ 无论是市面上主流的 rpc 框架还是小众的 rpc 框架都实现了上述 rpc的语义。【服务治理型:dubbo、dubbox、motan;多语言型:grpc、thrift、avro、protocol buffers】

打一波广告:【博主最近在写一个 java 实现的 rpc 框架 bridge 欢迎关注,考虑Mesh 化】

一、原理

首先用一幅图来简单描述一下 rpc 的调用过程,从 dubbo 官网拿来的,不算是最简单的图,但是也非常简单了,去掉上面的 Registry 和下面的 Monitor 剩下的就是最简单的 rpc 调用,说白了就是一个网络请求。

过程描述:

  1. 启动服务端provider,并向注册中心登记一下自己暴露服务的地址和服务详情
  2. 然后启动消费端consumer, 订阅注册中心的内容,也就是订阅服务,获取服务的详情
  3. 如果服务有变动,注册中心会通知消费端去更新订阅内容,更新服务详情。
  4. 客户端拿到了服务详情,通过网络对服务端发起网络请求,获取结果
  5. 监视器可以获取到服务调用详情和消费详情,但不限于此

OK,原理就是这么简单,接下来根据上面的描述逐步实现。

二、动手实践

下面基于 springboot 来实现上述的过程。

2.1 构建模块

搭建工程和子模块,工程结构如下:

2.2 实现服务端

看下服务端的内容,贴图

把接口定义在 api 模块,consumer 和 provider 模块都要引用到,接口HelloService代码如下

package com.glmapper.simple.api;

/**
 * service interface
 *
 * @author: Jerry
 */
public interface HelloService {

    /**
     * service function
     *
     * @param name
     * @return
     */
    String hello(String name);
}
复制代码

然后在 provider 模块实现接口,用自定注解 @SimpleProvider 标识,先看下注解内容

package com.glmapper.simple.provider.annotation;

/**
 * 自定义服务注解
 *
 * @author Jerry
 */
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
// 标明可被 Spring 扫描
@Component 
public @interface SimpleProvider {

    Class<?> value();
}
复制代码

注解使用了@Component标识,所以可被 spring 扫描到,接下来看实现类HelloServiceImpl

package com.glmapper.simple.provider.service;

/**
 * service implement class
 *
 * @author: Jerry
 */
@SimpleProvider(HelloService.class)
public class HelloServiceImpl implements HelloService {

    /**
     * service function
     *
     * @param name
     * @return
     */
    @Override
    public String hello(String name) {
        return "Hello! " + name;
    }
}
复制代码

在定义一个服务配置的类SimpleProviderProperties,方便通过 application.yml 文件配置,

package com.glmapper.simple.provider.property;

/**
 * provider properties
 *
 * @author: Jerry
 */
public class SimpleProviderProperties {

    /**
     * 暴露服务的端口
     */
    private Integer port;

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }
}
复制代码

到这里基础的类文件就已经结束了,下面开始服务初始化,入口 ProviderInitializer

package com.glmapper.simple.provider;

/**
 * 启动并注册服务
 *
 * @author Jerry
 */
public class ProviderInitializer implements ApplicationContextAware, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProviderInitializer.class);

    private SimpleProviderProperties providerProperties;

    /**
     * service registry
     */
    private ServiceRegistry serviceRegistry;

    /**
     * store interface and service implement mapping
     */
    private Map<String, Object> handlerMap = new HashMap<>();

    public ProviderInitializer(SimpleProviderProperties providerProperties, ServiceRegistry serviceRegistry) {
        this.providerProperties = providerProperties;
        this.serviceRegistry = serviceRegistry;
    }

    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        // 获取被 SimpleProvider 注解的 Bean
        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(SimpleProvider.class);
        if (MapUtils.isNotEmpty(serviceBeanMap)) {
            for (Object serviceBean : serviceBeanMap.values()) {
                String interfaceName = serviceBean.getClass().getAnnotation(SimpleProvider.class).value().getName();
                handlerMap.put(interfaceName, serviceBean);
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            ChannelHandler channelHandler = new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel channel) throws Exception {
                    channel.pipeline()
                            .addLast(new SimpleDecoder(SimpleRequest.class))
                            .addLast(new SimpleEncoder(SimpleResponse.class))
                            .addLast(new SimpleHandler(handlerMap));
                }
            };
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(channelHandler)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            String host = getLocalHost();
            if (null == host) {
                LOGGER.error("can't get service address,because address is null");
                throw new SimpleException("can't get service address,because address is null");
            }
            int port = providerProperties.getPort();
            ChannelFuture future = bootstrap.bind(host, port).sync();
            LOGGER.debug("server started on port {}", port);

            if (serviceRegistry != null) {
                String serverAddress = host + ":" + port;
                serviceRegistry.register(serverAddress);
            }
            future.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    /**
     * get service host
     *
     * @return
     */
    private String getLocalHost() {
        Enumeration<NetworkInterface> allNetInterfaces;
        try {
            allNetInterfaces = NetworkInterface.getNetworkInterfaces();
        } catch (SocketException e) {
            LOGGER.error("get local address error,cause:", e);
            return null;
        }
        while (allNetInterfaces.hasMoreElements()) {
            NetworkInterface netInterface = allNetInterfaces.nextElement();
            Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
            while (addresses.hasMoreElements()) {
                InetAddress ip = addresses.nextElement();
                if (ip instanceof Inet4Address && !ip.isLoopbackAddress() && !ip.getHostAddress().contains(":")) {
                    return ip.getHostAddress();
                }
            }
        }
        return null;
    }
}
复制代码

描述一下这个类做了什么工作:

  • 首先他实现了ApplicationContextAware, InitializingBean这两个 spring 中接口,根据IOC容器初始化的顺序,会依次回调用接口中的setApplicationContextafterPropertiesSet 方法。
    • setApplicationContext方法中获取了容器中被@SimpleProvider标注的类,并将服务接口名和服务实现类绑定,存放到handlerMap中,在@SimpleProvider中有一个 value 属性,是考虑到一个类可以实现多个接口,通过 value 可以指定哪个服务接口,当然也可以定义为数组,处理多个接口
    • afterPropertiesSet 方法中做了两件事:
      • 在服务端开启了一个处理socket请求的线程池,监听和处理服务暴露端口上接受到的请求,指定了一个处理器SimpleHandler
      • 调用ServiceRegistry类的registry方法向 zookeeper 注册服务的地址和端口,这里没有用到协议,只注册了 ip:port

SimpleHandler是一个实现了 nettySimpleChannelInboundHandler的请求处理器类

package com.glmapper.simple.provider.handler;

/**
 * request handler
 *
 * @author Jerry
 */
public class SimpleHandler extends SimpleChannelInboundHandler<SimpleRequest> {

    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleHandler.class);

    private final Map<String, Object> handlerMap;

    public SimpleHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, SimpleRequest request) throws Exception {
        SimpleResponse response = new SimpleResponse();
        response.setRequestId(request.getRequestId());
        try {
            Object result = handle(request);
            response.setResult(result);
        } catch (Throwable t) {
            response.setError(t);
        }
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }

    private Object handle(SimpleRequest request) throws Throwable {
        String className = request.getClassName();
        Object serviceBean = handlerMap.get(className);

        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();

        FastClass serviceFastClass = FastClass.create(serviceClass);
        FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
        return serviceFastMethod.invoke(serviceBean, parameters);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.error("server caught exception", cause);
        ctx.close();
    }
}
复制代码

SimpleHandler基于 netty 的事件驱动模型触发对应的方法,当收到请求事件会调用channelRead0方法,这个方法的作用就是,根据请求参数中的接口名找到对应的实现类调用指定的方法,然后把结果返回。

再瞅瞅ServiceRegistry,入口是ProviderInitializer调用了ServiceRegistryregistry方法

package com.glmapper.simple.provider.registry;

/**
 * connect zookeeper to registry service
 *
 * @author Jerry
 */
public class ServiceRegistry {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceRegistry.class);

    private ZookeeperProperties zookeeperProperties;

    public ServiceRegistry(ZookeeperProperties zookeeperProperties) {
        this.zookeeperProperties = zookeeperProperties;
    }

    public void register(String data) {
        if (data != null) {
            ZooKeeper zk = ZookeeperUtils.connectServer(zookeeperProperties.getAddress(), zookeeperProperties.getTimeout());
            if (zk != null) {
                addRootNode(zk);
                createNode(zk, data);
            }
        }
    }

    /**
     * add one zookeeper root node
     *
     * @param zk
     */
    private void addRootNode(ZooKeeper zk) {
        try {
            String registryPath = zookeeperProperties.getRootPath();
            Stat s = zk.exists(registryPath, false);
            if (s == null) {
                zk.create(registryPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (KeeperException | InterruptedException e) {
            LOGGER.error("zookeeper add root node error,cause:", e);
        }
    }

    private void createNode(ZooKeeper zk, String data) {
        try {
            byte[] bytes = data.getBytes(Charset.forName("UTF-8"));
            String dataPath = zookeeperProperties.getRootPath() + zookeeperProperties.getDataPath();
            String path = zk.create(dataPath, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            LOGGER.debug("create zookeeper node ({} => {})", path, data);
        } catch (KeeperException | InterruptedException e) {
            LOGGER.error("create zookeeper node error,cause:", e);
        }
    }
}
复制代码

ServiceRegistry类做的工作比较简单,就是把 服务ip:port注册到 zk 的指定目录下

  • 创建根节点,根节点是个永久节点
  • 在根节点下创建临时的子节点,子节点存储了服务的 ip:port,服务被挂掉对应的子节点就会被干掉

2.3 消费端

消费端内容:

消费端的内容比较少,核心就三个类:ServiceDiscoveryConsumerHandlerConsumerProxy

先看下ServiceDiscovery内容:

package com.glmapper.simple.consumer.discovery;

/**
 * 服务发现:连接ZK,添加watch事件
 *
 * @author Jerry
 */
public class ServiceDiscovery {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceDiscovery.class);

    private volatile List<String> nodes = new ArrayList<>();

    private ZookeeperProperties zookeeperProperties;

    public ServiceDiscovery(ZookeeperProperties zookeeperProperties) {
        this.zookeeperProperties = zookeeperProperties;
        String address = zookeeperProperties.getAddress();
        int timeout = zookeeperProperties.getTimeout();
        ZooKeeper zk = ZookeeperUtils.connectServer(address, timeout);
        if (zk != null) {
            watchNode(zk);
        }
    }

    public String discover() {
        String data = null;
        int size = nodes.size();
        if (size > 0) {
            if (size == 1) {
                data = nodes.get(0);
                LOGGER.debug("using only node: {}", data);
            } else {
                data = nodes.get(ThreadLocalRandom.current().nextInt(size));
                LOGGER.debug("using random node: {}", data);
            }
        }
        return data;
    }

    private void watchNode(final ZooKeeper zk) {
        try {
            Watcher childrenNodeChangeWatcher = event -> {
                if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                    watchNode(zk);
                }
            };
            String rootPath = zookeeperProperties.getRootPath();
            List<String> nodeList = zk.getChildren(rootPath, childrenNodeChangeWatcher);
            List<String> nodes = new ArrayList<>();
            for (String node : nodeList) {
                byte[] bytes = zk.getData(rootPath + "/" + node, false, null);
                nodes.add(new String(bytes, Charset.forName("UTF-8")));
            }
            LOGGER.info("node data: {}", nodes);
            this.nodes = nodes;
        } catch (KeeperException | InterruptedException e) {
            LOGGER.error("节点监控出错,原因:", e);
        }
    }
}
复制代码

这个类的入口是构造器,作用是获取 zk 的地址,然后获取 zk 上的节点信息,这里没有实现服务订阅,也就是说如果 zk 上原本有两个服务,挂掉一个,客户端不会剔除挂掉的服务信息,导致调用失败。

然后是ConsumerProxy,它是一个代理工厂:

package com.glmapper.simple.consumer.proxy;

/**
 * ConsumerProxy
 *
 * @author Jerry
 */
public class ConsumerProxy {

    private ServiceDiscovery serviceDiscovery;

    public ConsumerProxy(ServiceDiscovery serviceDiscovery) {
        this.serviceDiscovery = serviceDiscovery;
    }

    @SuppressWarnings("unchecked")
    public <T> T create(Class<?> interfaceClass) {
        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
                new Class<?>[]{interfaceClass},
                new SimpleInvocationHandler());
    }

    private class SimpleInvocationHandler implements InvocationHandler {
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            SimpleRequest request = buildRequest(method, args);
            String serverAddress = getServerAddress();
            String[] array = serverAddress.split(":");
            String host = array[0];
            int port = Integer.parseInt(array[1]);
            ConsumerHandler consumerHandler = new ConsumerHandler(host, port);
            SimpleResponse response = consumerHandler.send(request);
            if (response.getError() != null) {
                throw new SimpleException("service invoker error,cause:", response.getError());
            } else {
                return response.getResult();
            }
        }

        private SimpleRequest buildRequest(Method method, Object[] args) {
            SimpleRequest request = new SimpleRequest();
            request.setRequestId(UUID.randomUUID().toString());
            request.setClassName(method.getDeclaringClass().getName());
            request.setMethodName(method.getName());
            request.setParameterTypes(method.getParameterTypes());
            request.setParameters(args);
            return request;
        }

        private String getServerAddress() {
            String serverAddress = null;
            if (serviceDiscovery != null) {
                serverAddress = serviceDiscovery.discover();
            }
            if (null == serverAddress) {
                throw new SimpleException("no server address available");
            }
            return serverAddress;
        }
    }
}
复制代码

这里有个内部类SimpleInvocationHandler是生产代理的核心,方法的核心是在 SimpleInvocationHandler.invoke()中是调用这两行代码

ConsumerHandler consumerHandler = new ConsumerHandler(host, port);
SimpleResponse response = consumerHandler.send(request);
复制代码

发起网络请求,下面看下ConsumerHandler

package com.glmapper.simple.consumer.handler;

/**
 * RPC真正调用客户端
 *
 * @author Jerry
 */
public class ConsumerHandler extends SimpleChannelInboundHandler<SimpleResponse> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerHandler.class);

    private int port;

    private String host;

    private SimpleResponse response;

    private CountDownLatch latch = new CountDownLatch(1);

    public ConsumerHandler(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, SimpleResponse response) throws Exception {
        this.response = response;
        latch.countDown();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.error("client caught exception", cause);
        ctx.close();
    }

    public SimpleResponse send(SimpleRequest request) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            ChannelInitializer<SocketChannel> channelHandler = new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel channel) throws Exception {
                    channel.pipeline()
                            // 将 RPC 请求进行编码(为了发送请求)
                            .addLast(new SimpleEncoder(SimpleRequest.class))
                            // 将 RPC 响应进行解码(为了处理响应)
                            .addLast(new SimpleDecoder(SimpleResponse.class))
                            // 使用 RpcClient 发送 RPC 请求
                            .addLast(ConsumerHandler.this);
                }
            };
            bootstrap.group(group).channel(NioSocketChannel.class)
                    .handler(channelHandler)
                    .option(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture future = bootstrap.connect(host, port).sync();
            future.channel().writeAndFlush(request).sync();
            latch.await();
            if (response != null) {
                future.channel().closeFuture().sync();
            }
            return response;
        } finally {
            group.shutdownGracefully();
        }
    }
}
复制代码

这个类和服务端的 ProviderHandler 的代码差不多,也是netty通讯类

附一下 GitHub 地址 simple-rpc

关注下面的标签,发现更多相似文章
评论