徒手编写服务发现框架系列二：服务提供端实现

﹏ヽ暗。殇╰゛Y 2022-11-27 08:48 35阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI3Mzc2NzM_size_16_color_FFFFFF_t_70][]

上一篇文章我们简单介绍了项目的背景及基本组件情况，这篇文章我们主要分享一下服务提供端的设计实现。在服务发现框架中，服务提供端的作用是为消费端提供接口调用，提供端应该首先启动，依次完成服务注册、接口暴露等等。

首先看下提供端的配置信息：

/**
     * @Author zxm
     * @Description
     * @Date Create in 下午 4:33 2019/1/22 0022
     */
    public class RegistryConfig {
        /**
         * 注册中心host
         */
        private String registryHost;
    
        /**
         * 注册中心端口
         */
        private Integer registryPort;
    
        /**
         * 应用名称
         */
        private String serverName;
    
        /**
         * 服务提供方端口
         */
        private Integer port;
    
        /**
         * 服务提供接口集合
         */
        private Set<ProviderConfig> providerConfigs;
    
        public String getRegistryHost() {
            return registryHost;
        }
    
        public void setRegistryHost(String registryHost) {
            this.registryHost = registryHost;
        }
    
        public Integer getRegistryPort() {
            return registryPort;
        }
    
        public void setRegistryPort(Integer registryPort) {
            this.registryPort = registryPort;
        }
    
        public String getServerName() {
            return serverName;
        }
    
        public void setServerName(String serverName) {
            this.serverName = serverName;
        }
    
        public Integer getPort() {
            return port;
        }
    
        public void setPort(Integer port) {
            this.port = port;
        }
    
        public Set<ProviderConfig> getProviderConfigs() {
            return providerConfigs;
        }
    
        public void setProviderConfigs(Set<ProviderConfig> providerConfigs) {
            this.providerConfigs = providerConfigs;
        }
    
        public String getLocalHostAddrss(){
            try {
                return InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                return null;
            }
        }
    }

RegistryConfig 为注册业务相关的配置信息，主要需要配置注册中心的ip和端口、服务的名称、接口提供方的端口以及暴漏的接口信息等等。

/**
     * @Author zxm
     * @Description
     * @Date Create in 上午 10:47 2019/1/23 0023
     */
    public class ProviderConfig<T>{
        /**
         * 接口全称
         */
        private String apiName;
    
        /**
         * 接口实现类
         */
        private Class<T> implementsClass;
    
        public String getApiName() {
            return apiName;
        }
    
        public void setApiName(String apiName) {
            this.apiName = apiName;
        }
    
        public Class<T> getImplementsClass() {
            return implementsClass;
        }
    
        public void setImplementsClass(Class<T> implementsClass) {
            this.implementsClass = implementsClass;
        }
    }

ProviderConfig是提供方对外暴露的接口的配置信息，包括接口的全限定名和实现类信息，用于通信过程中被反射调用。

注册处理器

/**
     * @Author zxm
     * @Description
     * @Date Create in 下午 4:48 2019/1/22 0022
     */
    public class RegistryHandler {
        private RegistryConfig registryConfig;
    
        private final RegistryClient registryClient;
    
        public RegistryHandler(RegistryConfig registryConfig) {
            this.registryConfig = registryConfig;
            this.registryClient = new RegistryClient(this.registryConfig);
        }
    
        public boolean registry() {
            return registryClient.registry();
        }
    
        public String getImplementClass(String apiName) {
            return registryClient.getImplementClass(apiName);
        }
    }

注册处理器RegistryHandler的作用是帮助服务提供端完成向注册中心的注册工作，初始化RegistryHandler时，会先初始化注册客户端RegistryClient，调用registry方法，将服务提供端的信息注册到注册中心（redis）中。

/**
     * @Author zxm
     * @Description
     * @Date Create in 下午 4:51 2019/1/22 0022
     */
    public class RegistryClient {
        private RegistryConfig registryConfig;
    
        private JedisTemplate template;
    
        public RegistryClient(RegistryConfig registryConfig) {
            this.registryConfig = registryConfig;
            template = JedisTemplateFactory.newInstance(registryConfig.getRegistryHost(), registryConfig.getRegistryPort());
        }
    
        public boolean registry(){
            String key = template.joinKey(registryConfig.getServerName(), registryConfig.getLocalHostAddrss(), registryConfig.getPort());
            String result = template.setValue(key, buildRegistryValue(registryConfig.getProviderConfigs()));
    
            if (null != result) {
                return true;
            }
            return false;
        }
    
        private String buildRegistryValue(Set<ProviderConfig> providerConfigs) {
            JSONObject value = new JSONObject();
            providerConfigs.forEach(providerConfig -> value.put(providerConfig.getApiName(), providerConfig.getImplementsClass()));
            return value.toJSONString();
        }
    
        /**
         * 根据apiName获取实现类
         * @param apiName
         * @return
         * @throws UnknownHostException
         */
        public String getImplementClass(String apiName){
            String key = template.joinKey(registryConfig.getServerName(), registryConfig.getLocalHostAddrss(), registryConfig.getPort());
            String result = template.getValue(key);
            return JSONObject.parseObject(result).get(apiName).toString();
        }
    }

注册的基本原理就是将提供者的接口信息保存在redis中，消费端启动的时候，访问redis数据，获取提供者的ip及接口信息列表。

至此，注册工作基本完成！

我们的框架的工作一个是服务发现，另一个就是实现rpc通信，提供端暴露完接口后，就要随时等待消费端的调用，下面我们看一下提供端是如何处理rpc调用逻辑的。

/**
     * @Author zxm
     * @Description zxm
     * @Date Create in 下午 2:53 2019/1/28 0028
     */
    public class ProviderProxyInvoker {
        private RegistryConfig registryConfig;
    
        private RegistryHandler registryHandler;
    
        public ProviderProxyInvoker(RegistryConfig registryConfig, RegistryHandler registryHandler) {
            if (null == registryConfig) {
                throw new RuntimeException("registryConfig is not null");
            }
            this.registryConfig = registryConfig;
            this.registryHandler = registryHandler;
            init(this.registryConfig);
        }
    
        /**
         * 异步初始化nettyServer
         * @param registryConfig
         */
        private void init(RegistryConfig registryConfig) {
            try {
                new Thread(new NettySocketServer(registryConfig.getPort(), this)).start();
            } catch (Exception e) {
                throw new RuntimeException(e.getMessage());
            }
        }
    
        /**
         * 反射调用实现类
         * @param rpcProtocol
         * @return
         */
        public Object invoke(RpcProtocol rpcProtocol) {
            try {
                Class implementClass = Class.forName(registryHandler.getImplementClass(rpcProtocol.getInterfaceName()));
                Method method = Class.forName(rpcProtocol.getInterfaceName())
                        .getMethod(rpcProtocol.getMethodName(), rpcProtocol.getParameterTypes());
                return method.invoke(implementClass.newInstance(), rpcProtocol.getArgs());
            } catch (ClassNotFoundException ex1) {
                throw new RuntimeException("api " + rpcProtocol.getInterfaceName() + " is not found, msg is:" + ex1.getMessage());
            } catch (NoSuchMethodException ex2) {
                throw new RuntimeException("method " + rpcProtocol.getMethodName() + " is not found");
            } catch (IllegalAccessException e) {
                return null;
            } catch (InvocationTargetException e) {
                return null;
            } catch (InstantiationException e) {
                return null;
            }
        }
    }

ProviderProxyInvoker作为提供端最核心的组件，主要完成网络通信层（nettyServer）的初始化以及反射调用接口实现类的工作。在初始化ProviderProxyInvoker时，会先初始化netty的server端，等待客户端的请求。

/**
         * 异步初始化nettyServer
         * @param registryConfig
         */
        private void init(RegistryConfig registryConfig) {
            try {
                new Thread(new NettySocketServer(registryConfig.getPort(), this)).start();
            } catch (Exception e) {
                throw new RuntimeException(e.getMessage());
            }
        }

我们的rpc通信协议主要是基于tcp的，因此我们底层采用netty实现，看一下初始化逻辑：

/**
     * @Author zxm
     * @Description netty服务端
     * @Date Create in 下午 2:19 2018/6/15 0015
     */
    @ChannelHandler.Sharable
    public class NettySocketServer implements Runnable{
        private static final Log log = LogFactory.getLog(NettySocketServer.class);
    
        private int port;
    
        private ProviderProxyInvoker invoker;
    
        public NettySocketServer(Integer port, ProviderProxyInvoker invoker) throws Exception {
            if (null == port) {
                this.port = 8888;
            } else {
                this.port = port.intValue();
            }
    
            if (invoker != null) {
                this.invoker = invoker;
            }
        }
    
        @Override
        public void run() {
            try {
                this.init(port);
            } catch (Exception e) {
            }
        }
    
        private void init(int port) throws Exception {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        //保持连接数
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        //有数据立即发送
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        //保持长连接
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) {
                                socketChannel
                                        .pipeline()
                                        .addLast(new StringDecoder(), new StringEncoder(), new NettyServerReadHandler(invoker));
                            }
                        });
    
                ChannelFuture f = bootstrap.bind(port).sync();
    
                if (f.isSuccess()) {
                    log.info("netty server started success!!!\r\n（Copyright© Nicholas.Tony)");
                } else {
                    log.info("long connection started fail");
                }
    
    
                f.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

/**
     * @Author zxm
     * @Description 服务端数据读取处理器
     * @Date Create in 下午 2:37 2018/6/15 0015
     */
    public class NettyServerReadHandler extends SimpleChannelInboundHandler<String> {
        Logger log = LoggerFactory.getLogger(NettyServerReadHandler.class);
    
        private ProviderProxyInvoker invoker;
    
        public NettyServerReadHandler(ProviderProxyInvoker invoker) {
            if (invoker != null) {
                this.invoker = invoker;
            }
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            channelReadExecute(ctx.channel(), msg);
        }
    
        private void channelReadExecute(Channel channel, String msg) {
            log.info("remote [{}] request success, params is:{}", channel.remoteAddress(), msg);
            NettyServerDispatcher.threadPool.submit(new NettyServerDispatcherHandler(invoker, channel, msg));
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            ctx.close();
        }
    
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            super.channelRegistered(ctx);
            log.info("client " + ctx.channel().remoteAddress() + " connected");
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
        }
    }

消息监听模块，采用线程池异步进行处理，提高并发能力，实际处理请求的处理器是NettyServerDispatcherHandler组件，将客户端的请求报文解析后，在提供端本地利用java反射机制，完成接口的调用工作，然后将调用结果同步返回给netty客户端。

/**
     * @Author zxm
     * @Description
     * @Date Create in 下午 3:34 2019/1/30 0030
     */
    public class NettyServerDispatcherHandler implements Runnable {
        private ProviderProxyInvoker invoker;
    
        private Channel channel;
        private String msg;
    
        public NettyServerDispatcherHandler(ProviderProxyInvoker invoker, Channel channel, String msg) {
            this.invoker = invoker;
            this.channel = channel;
            this.msg = msg;
        }
    
        @Override
        public void run() {
            RpcProtocol rpcProtocol = RpcSerializer.deSerialize(msg, RpcProtocol.class);
            Object response = invoker.invoke(rpcProtocol);
            channel.writeAndFlush(RpcSerializer.serialize(rpcResultWrapper(rpcProtocol.getId(), response)));
        }
    
        private RpcResult rpcResultWrapper(String id, Object response) {
            RpcResult rpcResult = new RpcResult();
            rpcResult.setId(id);
            rpcResult.setValue(response);
            return rpcResult;
        }
    }

至此提供端rpc通信模块的相关组件逻辑的实现基本完成，总结下来，原理也比较简单。在下一篇中我们会继续分享消费端的相关实现。

项目地址： [https://github.com/zhangxiaomin1993/rpc-server-sdk][https_github.com_zhangxiaomin1993_rpc-server-sdk]

声明：文章和项目不以商业盈利为目的，仅作为本人技术积累的沉淀，分享给大家，有兴趣的朋友欢迎访问交流，共同学习和进步！大佬和专家路过，不喜勿喷！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI3Mzc2NzM_size_16_color_FFFFFF_t_70]: /images/20221124/e0d752551f0442cb8bf5b3bacc3fa3bf.png
[https_github.com_zhangxiaomin1993_rpc-server-sdk]: https://github.com/zhangxiaomin1993/rpc-server-sdk