自定义实现一个RPC框架（六）

左手的ㄟ右手 2023-02-13 11:52 51阅读 0赞

### 文章目录 ###

*  自定义实现一个RPC框架（五）：
 *   *   *  地址：https://gitee.com/liyunfengfengfeng/rpc
         *  基于netty来实现

# 自定义实现一个RPC框架（五）： #

### 地址：https://gitee.com/liyunfengfengfeng/rpc ###

### 基于netty来实现 ###

package com.luban.protocol.dubbo;
    
    import com.luban.framework.Invocation;
    import com.luban.framework.Protocol;
    import com.luban.framework.URL;
    
    public class DubboProtocol implements Protocol { 
    
        @Override
        public void start(URL url) { 
            NettyServer nettyServer = new NettyServer();
            nettyServer.start(url.getHostname(), url.getPort());
    
        }
    
        @Override
        public String send(URL url, Invocation invocation) { 
            NettyClient nettyClient = new NettyClient();
            return nettyClient.send(url.getHostname(),url.getPort(), invocation);
        }
    }

package com.luban.protocol.dubbo;
    
    import com.luban.framework.Invocation;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.serialization.ClassResolvers;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class NettyClient<T> { 
    
        public NettyClientHandler client = null;
    
        private static ExecutorService executorService = Executors.newCachedThreadPool();
    
        public void start(String hostName, Integer port) { 
            client = new NettyClientHandler();
    
            // 服务器线程组 用于网络事件的处理 一个用于服务器接收客户端的连接
            // 另一个线程组用于处理SocketChannel的网络读写
            Bootstrap b = new Bootstrap();
            EventLoopGroup group = new NioEventLoopGroup();
            b.group(group)
                    .channel(NioSocketChannel.class)//类似NIO中serverSocketChannel
                    .option(ChannelOption.TCP_NODELAY, true)// 配置TCP参数
                    .handler(new ChannelInitializer<SocketChannel>() { 
                        protected void initChannel(SocketChannel socketChannel) throws Exception { 
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //netty5.0的序列化与反序列化ObjectDecoder和ObjectEncoder
                            pipeline.addLast("decoder", new ObjectDecoder(ClassResolvers
                                    .weakCachingConcurrentResolver(this.getClass()
                                            .getClassLoader())));
                            pipeline.addLast("encoder", new ObjectEncoder());
                            pipeline.addLast("handler", client);
                        }
                    });
    
            try { 
                b.connect(hostName, port).sync();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public String send(String hostName, Integer port, Invocation invocation) { 
            if (client == null) { 
                start(hostName, port);
            }
    
            client.setInvocation(invocation);
            try { 
                return (String) executorService.submit(client).get();
            } catch (InterruptedException e) { 
    // e.printStackTrace();
            } catch (ExecutionException e) { 
    // e.printStackTrace();
            }
            return null;
        }
    
    
    }

package com.luban.protocol.dubbo;
    
    import com.luban.framework.Invocation;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.util.concurrent.Callable;
    
    public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable { 
    
        private ChannelHandlerContext context;
        private Invocation invocation;
        private String result;
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            context = ctx;
        }
    
        @Override
        public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            result = msg.toString();
            notify();
        }
    
        public synchronized Object call() throws Exception { 
            context.writeAndFlush(this.invocation);
            wait();
            return result;
        }
    
        public void setInvocation(Invocation invocation) { 
            this.invocation = invocation;
        }
    }

package com.luban.protocol.dubbo;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.serialization.ClassResolvers;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    
    public class NettyServer { 
    
        public void start(String hostName, int port) { 
            try { 
                final ServerBootstrap bootstrap = new ServerBootstrap();
                NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
                bootstrap.group(eventLoopGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
    
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                pipeline.addLast("decoder", new ObjectDecoder(ClassResolvers
                                        .weakCachingConcurrentResolver(this.getClass()
                                                .getClassLoader())));
                                pipeline.addLast("encoder", new ObjectEncoder());
                                pipeline.addLast("handler", new NettyServerHandler());
                            }
                        });
                bootstrap.bind(hostName, port).sync();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public void close() { 
    
        }
    }

package com.luban.protocol.dubbo;
    
    import com.luban.framework.Invocation;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import com.luban.provider.LocalRegister;
    
    import java.lang.reflect.Method;
    
    public class NettyServerHandler extends ChannelInboundHandlerAdapter { 
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            Invocation invocation = (Invocation) msg;
    
            Class serviceImpl = LocalRegister.get(invocation.getInterfaceName());
    
            Method method = serviceImpl.getMethod(invocation.getMethodName(), invocation.getParamType());
            Object result = method.invoke(serviceImpl.newInstance(), invocation.getParams());
    
            System.out.println("Netty-------------" + result.toString());
            ctx.writeAndFlush("Netty:" + result);
        }
    }
    
           Object result = method.invoke(serviceImpl.newInstance(), invocation.getParams());
    
            System.out.println("Netty-------------" + result.toString());
            ctx.writeAndFlush("Netty:" + result);
        }
    }