由浅入深Netty简易实现RPC框架

叁歲伎倆 2024-03-17 09:13 20阅读 0赞

#### 目录 ####

*   *  1 准备工作
     *  2 服务器 handler
     *  3 客户端代码第一版
     *  4 客户端 handler 第一版
     *  5 客户端代码 第二版
     *  6 客户端 handler 第二版

--------------------

### 1 准备工作 ###

![在这里插入图片描述][19db2026515843eca5d2a95665a3d2b7.png]

这些代码可以认为是现成的，无需从头编写练习

为了简化起见，在原来聊天项目的基础上新增 Rpc 请求和响应消息

@Data
    public abstract class Message implements Serializable {
        
    
        // 省略旧的代码
    
        public static final int RPC_MESSAGE_TYPE_REQUEST = 101;
        public static final int  RPC_MESSAGE_TYPE_RESPONSE = 102;
    
        static {
        
            // ...
            messageClasses.put(RPC_MESSAGE_TYPE_REQUEST, RpcRequestMessage.class);
            messageClasses.put(RPC_MESSAGE_TYPE_RESPONSE, RpcResponseMessage.class);
        }
    
    }

请求消息

@Getter
    @ToString(callSuper = true)
    public class RpcRequestMessage extends Message {
        
    
        /**
         * 调用的接口全限定名，服务端根据它找到实现
         */
        private String interfaceName;
        /**
         * 调用接口中的方法名
         */
        private String methodName;
        /**
         * 方法返回类型
         */
        private Class<?> returnType;
        /**
         * 方法参数类型数组
         */
        private Class[] parameterTypes;
        /**
         * 方法参数值数组
         */
        private Object[] parameterValue;
    
        public RpcRequestMessage(int sequenceId, String interfaceName, String methodName, Class<?> returnType, Class[] parameterTypes, Object[] parameterValue) {
        
            super.setSequenceId(sequenceId);
            this.interfaceName = interfaceName;
            this.methodName = methodName;
            this.returnType = returnType;
            this.parameterTypes = parameterTypes;
            this.parameterValue = parameterValue;
        }
    
        @Override
        public int getMessageType() {
        
            return RPC_MESSAGE_TYPE_REQUEST;
        }
    }

响应消息

@Data
    @ToString(callSuper = true)
    public class RpcResponseMessage extends Message {
        
        /**
         * 返回值
         */
        private Object returnValue;
        /**
         * 异常值
         */
        private Exception exceptionValue;
    
        @Override
        public int getMessageType() {
        
            return RPC_MESSAGE_TYPE_RESPONSE;
        }
    }

服务器架子

@Slf4j
    public class RpcServer {
        
        public static void main(String[] args) {
        
            NioEventLoopGroup boss = new NioEventLoopGroup();
            NioEventLoopGroup worker = new NioEventLoopGroup();
            LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
            MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
            
            // rpc 请求消息处理器，待实现
            RpcRequestMessageHandler RPC_HANDLER = new RpcRequestMessageHandler();
            try {
        
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.channel(NioServerSocketChannel.class);
                serverBootstrap.group(boss, worker);
                serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
        
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
        
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
                        ch.pipeline().addLast(LOGGING_HANDLER);
                        ch.pipeline().addLast(MESSAGE_CODEC);
                        ch.pipeline().addLast(RPC_HANDLER);
                    }
                });
                Channel channel = serverBootstrap.bind(8080).sync().channel();
                channel.closeFuture().sync();
            } catch (InterruptedException e) {
        
                log.error("server error", e);
            } finally {
        
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        }
    }

客户端架子

public class RpcClient {
        
        public static void main(String[] args) {
        
            NioEventLoopGroup group = new NioEventLoopGroup();
            LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
            MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
            
            // rpc 响应消息处理器，待实现
            RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
            try {
        
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.group(group);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
        
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
                        ch.pipeline().addLast(LOGGING_HANDLER);
                        ch.pipeline().addLast(MESSAGE_CODEC);
                        ch.pipeline().addLast(RPC_HANDLER);
                    }
                });
                Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
                channel.closeFuture().sync();
            } catch (Exception e) {
        
                log.error("client error", e);
            } finally {
        
                group.shutdownGracefully();
            }
        }
    }

服务器端的 service 获取

public class ServicesFactory {
        
    
        static Properties properties;
        static Map<Class<?>, Object> map = new ConcurrentHashMap<>();
    
        static {
        
            try (InputStream in = Config.class.getResourceAsStream("/application.properties")) {
        
                properties = new Properties();
                properties.load(in);
                Set<String> names = properties.stringPropertyNames();
                for (String name : names) {
        
                    if (name.endsWith("Service")) {
        
                        Class<?> interfaceClass = Class.forName(name);
                        Class<?> instanceClass = Class.forName(properties.getProperty(name));
                        map.put(interfaceClass, instanceClass.newInstance());
                    }
                }
            } catch (IOException | ClassNotFoundException | InstantiationException | IllegalAccessException e) {
        
                throw new ExceptionInInitializerError(e);
            }
        }
    
        public static <T> T getService(Class<T> interfaceClass) {
        
            return (T) map.get(interfaceClass);
        }
    }

相关配置 application.properties

serializer.algorithm=Json
    cn.itcast.server.service.HelloService=cn.itcast.server.service.HelloServiceImpl

### 2 服务器 handler ###

@Slf4j
    @ChannelHandler.Sharable
    public class RpcRequestMessageHandler extends SimpleChannelInboundHandler<RpcRequestMessage> {
        
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RpcRequestMessage message) {
        
            RpcResponseMessage response = new RpcResponseMessage();
            response.setSequenceId(message.getSequenceId());
            try {
        
                // 获取真正的实现对象
                HelloService service = (HelloService)
                        ServicesFactory.getService(Class.forName(message.getInterfaceName()));
                
                // 获取要调用的方法
                Method method = service.getClass().getMethod(message.getMethodName(), message.getParameterTypes());
                
                // 调用方法
                Object invoke = method.invoke(service, message.getParameterValue());
                // 调用成功
                response.setReturnValue(invoke);
            } catch (Exception e) {
        
                e.printStackTrace();
                // 调用异常
                response.setExceptionValue(e);
            }
            // 返回结果
            ctx.writeAndFlush(response);
        }
    }

### 3 客户端代码第一版 ###

只发消息

@Slf4j
    public class RpcClient {
        
        public static void main(String[] args) {
        
            NioEventLoopGroup group = new NioEventLoopGroup();
            LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
            MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
            RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
            try {
        
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.group(group);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
        
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
                        ch.pipeline().addLast(LOGGING_HANDLER);
                        ch.pipeline().addLast(MESSAGE_CODEC);
                        ch.pipeline().addLast(RPC_HANDLER);
                    }
                });
                Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
    
                ChannelFuture future = channel.writeAndFlush(new RpcRequestMessage(
                        1,
                        "cn.itcast.server.service.HelloService",
                        "sayHello",
                        String.class,
                        new Class[]{
        String.class},
                        new Object[]{
        "张三"}
                )).addListener(promise -> {
        
                    if (!promise.isSuccess()) {
        
                        Throwable cause = promise.cause();
                        log.error("error", cause);
                    }
                });
    
                channel.closeFuture().sync();
            } catch (Exception e) {
        
                log.error("client error", e);
            } finally {
        
                group.shutdownGracefully();
            }
        }
    }

### 4 客户端 handler 第一版 ###

@Slf4j
    @ChannelHandler.Sharable
    public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
        
            log.debug("{}", msg);
        }
    }

### 5 客户端代码 第二版 ###

包括 channel 管理，代理，接收结果

@Slf4j
    public class RpcClientManager {
        
    
    
        public static void main(String[] args) {
        
            HelloService service = getProxyService(HelloService.class);
            System.out.println(service.sayHello("zhangsan"));
    //        System.out.println(service.sayHello("lisi"));
    //        System.out.println(service.sayHello("wangwu"));
        }
    
        // 创建代理类
        public static <T> T getProxyService(Class<T> serviceClass) {
        
            ClassLoader loader = serviceClass.getClassLoader();
            Class<?>[] interfaces = new Class[]{
        serviceClass};
            //                                                            sayHello  "张三"
            Object o = Proxy.newProxyInstance(loader, interfaces, (proxy, method, args) -> {
        
                // 1. 将方法调用转换为 消息对象
                int sequenceId = SequenceIdGenerator.nextId();
                RpcRequestMessage msg = new RpcRequestMessage(
                        sequenceId,
                        serviceClass.getName(),
                        method.getName(),
                        method.getReturnType(),
                        method.getParameterTypes(),
                        args
                );
                // 2. 将消息对象发送出去
                getChannel().writeAndFlush(msg);
    
                // 3. 准备一个空 Promise 对象，来接收结果             指定 promise 对象异步接收结果线程
                DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
                RpcResponseMessageHandler.PROMISES.put(sequenceId, promise);
    
    //            promise.addListener(future -> {
        
    //                // 线程
    //            });
    
                // 4. 等待 promise 结果
                promise.await();
                if(promise.isSuccess()) {
        
                    // 调用正常
                    return promise.getNow();
                } else {
        
                    // 调用失败
                    throw new RuntimeException(promise.cause());
                }
            });
            return (T) o;
        }
    
        private static Channel channel = null;
        private static final Object LOCK = new Object();
    
        // 获取唯一的 channel 对象
        public static Channel getChannel() {
        
            if (channel != null) {
        
                return channel;
            }
            synchronized (LOCK) {
         //  t2
                if (channel != null) {
         // t1
                    return channel;
                }
                initChannel();
                return channel;
            }
        }
    
        // 初始化 channel 方法
        private static void initChannel() {
        
            NioEventLoopGroup group = new NioEventLoopGroup();
            LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
            MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
            RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
        
                    ch.pipeline().addLast(new ProcotolFrameDecoder());
                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    ch.pipeline().addLast(RPC_HANDLER);
                }
            });
            try {
        
                channel = bootstrap.connect("localhost", 8080).sync().channel();
                channel.closeFuture().addListener(future -> {
        
                    group.shutdownGracefully();
                });
            } catch (Exception e) {
        
                log.error("client error", e);
            }
        }
    }

### 6 客户端 handler 第二版 ###

@Slf4j
    @ChannelHandler.Sharable
    public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
        
    
        //                       序号      用来接收结果的 promise 对象
        public static final Map<Integer, Promise<Object>> PROMISES = new ConcurrentHashMap<>();
    
        @Override
    
        protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
        
            log.debug("{}", msg);
            // 拿到空的 promise
            Promise<Object> promise = PROMISES.remove(msg.getSequenceId());
            if (promise != null) {
        
                Object returnValue = msg.getReturnValue();
                Exception exceptionValue = msg.getExceptionValue();
                if(exceptionValue != null) {
        
                    promise.setFailure(exceptionValue);
                } else {
        
                    promise.setSuccess(returnValue);
                }
            }
        }
    }

[19db2026515843eca5d2a95665a3d2b7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/17/24b0aef0a39449d5acb12d85958afb3f.png