【Netty】二、利用Netty手写RPC远程调用

ゝ一世哀愁。 2023-10-09 15:02 39阅读 0赞

## 一、使用Netty实现标准案例 ##

需求：  
客户端发送now，服务端返回当前北京时间；  
客户端发送bye，客户端就退出；  
客户端发送其他命令，服务端返回：命令找不到；

### 客户端代码 ###

**
     *   标准事件案例 客户端向服务端发请求，服务端向客户端回应请求
     * 北京时间案例：
     * 客户端发送now，服务端返回当前北京时间；
     * 客户端发送bye，客户端就退出；
     * 客户端发送其他命令，服务端返回：命令找不到；
     */
    public class TimeClient {
        
    
        public static final String IP = "127.0.0.1";
    
        public static final int PORT = 8090;
    
        public static void main(String[] args) {
        
            TimeClient timeClient = new TimeClient();
            timeClient.run();
        }
    
        public void run() {
        
    
            //1、创建一个无限循环的线程池组
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try {
        
                //2、启动引导类 (客户端需要new Bootstrap)
                Bootstrap bootstrap = new Bootstrap();
    
                //3、给启动引导类做一些配置：
                // - NioServerSocketChannel
                // - childHandler
                // - bind端口
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class) // nio的socket通道
                        .handler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                ChannelPipeline channelPipeline = socketChannel.pipeline();
                                channelPipeline.addLast(new TimeClientHandler());
                            }
                        });
    
                //绑定一个端口，返回未来的通道
                ChannelFuture channelFuture = bootstrap.connect(IP, PORT).sync();
    
                //得到一个通道
                Channel channel = channelFuture.channel();
    
                //从控制台输入数据，往服务端发送
                Scanner scanner = new Scanner(System.in);
                for (;;) {
        
                    String line = scanner.nextLine();
                    if (line.equals("bye")) {
        
                        break;
                    }
    
                    //数据需要写到一个ByteBuf, 然后把ByteBuf写到服务端， now
                    channel.writeAndFlush(Unpooled.copiedBuffer(line, CharsetUtil.UTF_8));
                }
    
                //4、当channel被关闭的时候会通知此处关闭chanel（closeFuture方法）
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                eventLoopGroup.shutdownGracefully();
            }
        }
    }

import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /**
      客户端处事件
    */
    public class TimeClientHandler extends ChannelInboundHandlerAdapter {
        
    
        public TimeClientHandler() {
        
            super();
        }
    
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        
            super.channelRegistered(ctx);
        }
    
        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        
            super.channelUnregistered(ctx);
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            super.channelActive(ctx);
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        
            super.channelInactive(ctx);
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            ByteBuf byteBuf = (ByteBuf) msg;
            String time = byteBuf.toString(CharsetUtil.UTF_8);
            System.out.println("客户端接收到的时间北京时间：" + time);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        
            super.channelReadComplete(ctx);
        }
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        
            super.userEventTriggered(ctx, evt);
        }
    
        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        
            super.channelWritabilityChanged(ctx);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            super.exceptionCaught(ctx, cause);
        }
    
        @Override
        protected void ensureNotSharable() {
        
            super.ensureNotSharable();
        }
    
        @Override
        public boolean isSharable() {
        
            return super.isSharable();
        }
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        
            super.handlerAdded(ctx);
        }
    
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        
            super.handlerRemoved(ctx);
        }
    }

### 服务端代码 ###

/**
     *  * 北京时间案例：
     *  * 客户端发送now，服务端返回当前北京时间；
     *  * 客户端发送bye，客户端就退出；
     *  * 客户端发送其他命令，服务端返回：命令找不到；
     */
    public class TimeServer {
        
    
        public static final int PORT = 8090;
    
        public static void main(String[] args) {
        
            TimeServer timeServer = new TimeServer();
            timeServer.run();
        }
    
        public void run() {
        
            final TimeServerHandler timeServerHandler = new TimeServerHandler();
    
            //1、创建一个无限循环线程池组
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workGroup = new NioEventLoopGroup();
            try {
        
                //2、启动引导类
                ServerBootstrap bootstrap = new ServerBootstrap();
    
                //3、给启动引导类做一些配置：
                // - NioServerSocketChannel
                // - childHandler
                // - bind端口
                bootstrap.group(bossGroup, workGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                ChannelPipeline channelPipeline = socketChannel.pipeline();
                                channelPipeline.addLast(timeServerHandler);
                            }
                        });
    
                //绑定一个端口，返回未来的通道
                ChannelFuture channelFuture = bootstrap.bind(PORT).sync();
    
                //4、当channel被关闭的时候会通知此处关闭chanel（closeFuture方法）
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                bossGroup.shutdownGracefully();
                workGroup.shutdownGracefully();
            }
        }
    }

服务端处理事件

@ChannelHandler.Sharable
    public class TimeServerHandler extends ChannelInboundHandlerAdapter {
        
    
        public TimeServerHandler() {
        
            super();
        }
    
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        
            super.channelRegistered(ctx);
        }
    
        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        
            super.channelUnregistered(ctx);
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            super.channelActive(ctx);
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        
            super.channelInactive(ctx);
        }
    
        /**
         * 客户端发送数据到channel，触发该方法读取该数据
         *
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
    
            ByteBuf byteBuf = (ByteBuf) msg;
    
            String message = byteBuf.toString(CharsetUtil.UTF_8);
            System.out.println("接收到的消息：" + message);
    
            if (message.equals("now")) {
        
                String nowTime = DateFormat.getDateTimeInstance().format(new Date());
                ctx.write(Unpooled.copiedBuffer(nowTime, CharsetUtil.UTF_8));
            }
        }
    
    
        /**
         * 读取完的事件
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        
            ctx.flush();
        }
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        
            super.userEventTriggered(ctx, evt);
        }
    
        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        
            super.channelWritabilityChanged(ctx);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            super.exceptionCaught(ctx, cause);
        }
    
        @Override
        protected void ensureNotSharable() {
        
            super.ensureNotSharable();
        }
    
        @Override
        public boolean isSharable() {
        
            return super.isSharable();
        }
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        
            super.handlerAdded(ctx);
        }
    
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        
            super.handlerRemoved(ctx);
        }
    }

效果  
![在这里插入图片描述][3dcc315dfbb14ea1aa6353a799ac2ad5.png]  
![在这里插入图片描述][7f72ff78df3a473ab614049b0545f6e2.png]

## 二、RPC远程调用案例 ##

RPC远程调用案例  
需求：  
类似于Dubbo的rpc调用；  
服务消费者通过IP和端口直连调用服务提供者；  
基于服务接口进行调用；  
服务接口的方法支持多个参数；  
服务接口的方法支持多种类型的参数；

## 接口层 ##

**user类**

import java.io.Serializable;
    import java.util.Date;
    
    public class User implements Serializable {
        
    
        private int id;
    
        private String nick;
    
        private Date birthDay;
    
        public int getId() {
        
            return id;
        }
    
        public void setId(int id) {
        
            this.id = id;
        }
    
        public String getNick() {
        
            return nick;
        }
    
        public void setNick(String nick) {
        
            this.nick = nick;
        }
    
        public Date getBirthDay() {
        
            return birthDay;
        }
    
        public void setBirthDay(Date birthDay) {
        
            this.birthDay = birthDay;
        }
    
        @Override
        public String toString() {
        
            return "User{" +
                    "id=" + id +
                    ", nick='" + nick + '\'' +
                    ", birthDay=" + birthDay +
                    '}';
        }
    }

**服务接口**

import com.mytest.model.User;
    
    /**
     * 服务接口
     */
    public interface SomeService {
        
    
        public String doSomeThing(String thing);
    
        public User getUser(User user);
    }

### 服务提供者 ###

import com.mytest.server.RpcServer;
    
    /**
     * 服务提供者启动类  Netty Rpc d=调用 服务提供者启动类 对应的客户端是rpc-consumer
     *
     */
    public class RpcProvider {
        
    
        public static void main(String[] args) {
        
            //启动，在一个端口等待调用  RpcServer累类在netty-rpc
            RpcServer rpcServer = new RpcServer();
    
            rpcServer.start(new String[] {
        
                    "com.mytest.service.impl.SomeServiceImpl"
            });
        }
    }

接口实现类

import com.mytest.model.User;
    import com.mytest.service.SomeService;
    
    public class SomeServiceImpl implements SomeService {
        
    
        @Override
        public String doSomeThing(String thing) {
        
            return "I am do " + thing;
        }
    
        @Override
        public User getUser(User user) {
        
            user.setNick("小明");
            return user;
        }
    }

### 服务提供者 ###

import com.mytest.constant.Constants;
    import com.mytest.model.User;
    import com.mytest.proxy.RpcInvokeProxy;
    import com.mytest.service.SomeService;
    
    import java.net.InetAddress;
    import java.net.UnknownHostException;
    import java.util.Date;
    
    
    /**
     * Netty服务消费者 对应的提供者是是rpc-provider
     */
    public class RpcConsumer {
        
    
        public static void main(String[] args) throws UnknownHostException {
        
    
            //远程服务的ip和端口是多少
            String serviceAddress = InetAddress.getLocalHost().getHostAddress() + ":" + Constants.DEFAULT_PORT;
    
            // 动态代理 $Proxy@9835
            SomeService someService = RpcInvokeProxy.createProxy(SomeService.class, serviceAddress);
    
            String result = someService.doSomeThing("study.....");
            System.out.println("客户端调用结果：" + result);
    
            User user = new User();
            user.setNick("张三");
            user.setId(1236);
            user.setBirthDay(new Date());
            User uu = someService.getUser(user);
            System.out.println("客户端调用结果：" + uu);
        }
    }

### RPC实现 ###

#### **RpcServer类实现提供服务** ####

import com.mytest.constant.Constants;
    import com.mytest.server.handler.RpcServerHandler;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    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.NioServerSocketChannel;
    import io.netty.handler.codec.serialization.ClassResolvers;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    import java.net.InetAddress;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    public class RpcServer {
        
    
        // key ： com.mytest.service.SomeService
        // velue : new SomeServiceImpl();
        public Map<String, Object> serviceMap = new ConcurrentHashMap<>();
    
        public void start(String[] services) {
        
            try {
        
                // clazz = com.mytest.service.impl.SomeServiceImpl
                for (String clazz :  services) {
        
                    Class<?> implClass = Class.forName(clazz);
                    String interfaceName = implClass.getInterfaces()[0].getName();
                    serviceMap.put(interfaceName, implClass.newInstance());
                }
                //服务所在机器的ip:port地址
                String serviceAddress = InetAddress.getLocalHost().getHostAddress() + ":" + Constants.DEFAULT_PORT;
    
                //以下代码是实现netty的服务端
                final RpcServerHandler rpcServerHandler = new RpcServerHandler(serviceMap);
    
                //1、创建一个线程池
                EventLoopGroup boosGroup = new NioEventLoopGroup(1);
                EventLoopGroup workGroup = new NioEventLoopGroup();
    
                //2、启动引导类
                ServerBootstrap bootstrap = new ServerBootstrap();
    
                //3、给启动引导类做一些配置：
                // - NioServerSocketChannel
                // - childHandler
                // - bind端口
                bootstrap.group(boosGroup, workGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                ChannelPipeline channelPipeline = socketChannel.pipeline();
                                channelPipeline.addLast(new LoggingHandler(LogLevel.INFO));
                                channelPipeline.addLast(new ObjectEncoder()); //用的是netty自带的对象的编解码实现，也可以自己实现编解码
                                channelPipeline.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
                                channelPipeline.addLast(rpcServerHandler);
                            }
                        });
    
                // 127.0.0.1:6666
                String ip = serviceAddress.split(":")[0];
                String port = serviceAddress.split(":")[1];
    
                //绑定一个端口，返回未来的通道
                ChannelFuture channelFuture = bootstrap.bind(ip, Integer.parseInt(port)).sync();
    
                //4、当channel被关闭的时候会通知此处关闭chanel（closeFuture方法）
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
        
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }
    }
    
    
    
    
    
    import com.mytest.invoke.InvokeRequest;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    @ChannelHandler.Sharable
    public class RpcServerHandler extends ChannelInboundHandlerAdapter {
        
    
        public Map<String, Object> serviceMap = new ConcurrentHashMap<>();
    
        public RpcServerHandler(Map<String, Object> serviceMap) {
        
            this.serviceMap = serviceMap;
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            super.channelActive(ctx);
        }
    
        /**
         * 当channel有数据可以读了，触发该方法
         *
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            if (msg instanceof InvokeRequest) {
        
                InvokeRequest invokeRequest = (InvokeRequest) msg;
    
                // com.bjpowernode.service.impl.SomeServiceImpl
                Object implInstance = serviceMap.get(invokeRequest.getServiceName());
    
                //发起对 com.mytest.service.impl.SomeServiceImpl 的具体方法的调用
                Object result = implInstance.getClass()
                        .getMethod(invokeRequest.getMethodName(), invokeRequest.getParamTypes())
                        .invoke(implInstance, invokeRequest.getParamValues());
    
                //把结果告诉给调用者（调用者就是远程的客户端）
                ctx.write(result);
            }
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        
            ctx.flush();
            ctx.close();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            super.exceptionCaught(ctx, cause);
        }
    }

#### **RpcInvokeProxy实现rpc调用** ####

import com.mytest.invoke.InvokeRequest;
    import com.mytest.proxy.handler.RpcInvokeHandler;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    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.lang.reflect.Method;
    import java.lang.reflect.Proxy;
    
    public class RpcInvokeProxy {
        
    
        @SuppressWarnings("all")
        public static <T> T createProxy(Class<?> interfaceClass, String serviceAddress) {
        
            return (T)Proxy.newProxyInstance(interfaceClass.getClassLoader(),
                    new Class<?>[] {
        interfaceClass},
                    new MyInvocationHandler(interfaceClass, serviceAddress)
                    /*new InvocationHandler() {
                        @Override
                        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                            if (Object.class.equals(method.getDeclaringClass())) {
                                return method.invoke(this, args);
                            }
                            //在此处完成远程调用即可
                            return rpcInvoke(interfaceClass, method, args, serviceAddress);
                        }
                    }*/
            );
        }
    
        /**
         * 远程调用
         *
         * @param interfaceClass
         * @param method
         * @param args
         * @param serviceAddress
         * @return
         */
        public static Object rpcInvoke(Class<?> interfaceClass, Method method, Object[] args, String serviceAddress) {
        
            //建立netty连接，发起调用
    
            RpcInvokeHandler rpcInvokeHandler = new RpcInvokeHandler();
    
            //1、创建一个线程池
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try {
        
                //2、启动引导类 (客户端需要new Bootstrap)
                Bootstrap bootstrap = new Bootstrap();
    
                //3、给启动引导类做一些配置：
                // - NioServerSocketChannel
                // - childHandler
                // - bind端口
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                ChannelPipeline channelPipeline = socketChannel.pipeline();
                                channelPipeline.addLast(new ObjectEncoder()); //用的是netty自带的对象的编解码实现，也可以自己实现编解码
                                channelPipeline.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
                                channelPipeline.addLast(rpcInvokeHandler);
                            }
                        });
    
                String ip = serviceAddress.split(":")[0];
                String port = serviceAddress.split(":")[1];
    
                //绑定一个端口，返回未来的通道
                ChannelFuture channelFuture = bootstrap.connect(ip, Integer.parseInt(port)).sync();
    
                InvokeRequest invokeRequest = new InvokeRequest();
                invokeRequest.setServiceName(interfaceClass.getName()); //com.mytest.server.SomeService
                invokeRequest.setMethodName(method.getName());
                invokeRequest.setParamTypes(method.getParameterTypes());
                invokeRequest.setParamValues(args);
    
                //得到一个通道
                Channel channel = channelFuture.channel();
                channel.writeAndFlush(invokeRequest);
    
                //4、当channel被关闭的时候会通知此处关闭chanel（closeFuture方法）
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                eventLoopGroup.shutdownGracefully();
            }
            return rpcInvokeHandler.getResult();
        }
    }
    
    
    /**
     * 发起netty远程调用的方法
     */
    public class MyInvocationHandler implements InvocationHandler {
        
    
        /**
         * 实现类
         */
        private Class<?> interfaceClass;
    
        /**
         * 地址
         */
        private String serviceAddress;
    
        public MyInvocationHandler(Class<?> interfaceClass, String serviceAddress) {
        
            this.interfaceClass = interfaceClass;
            this.serviceAddress = serviceAddress;
        }
    
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        
            if (Object.class.equals(method.getDeclaringClass())) {
        
                return method.invoke(this, args);
            }
            //在此处完成远程调用即可
            return rpcInvoke(interfaceClass, method, args, serviceAddress);
        }
    
        /**
         * 远程调用
         *
         * @param interfaceClass
         * @param method
         * @param args
         * @param serviceAddress
         * @return
         */
        public static Object rpcInvoke(Class<?> interfaceClass, Method method, Object[] args, String serviceAddress) {
        
            //建立netty连接，发起调用
    
            RpcInvokeHandler rpcInvokeHandler = new RpcInvokeHandler();
    
            //1、创建一个线程池
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try {
        
                //2、启动引导类 (客户端需要new Bootstrap)
                Bootstrap bootstrap = new Bootstrap();
    
                //3、给启动引导类做一些配置：
                // - NioServerSocketChannel
                // - childHandler
                // - bind端口
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                ChannelPipeline channelPipeline = socketChannel.pipeline();
                                channelPipeline.addLast(new ObjectEncoder()); //用的是netty自带的对象的编解码实现，也可以自己实现编解码
                                channelPipeline.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
                                channelPipeline.addLast(rpcInvokeHandler);
                            }
                        });
    
                String ip = serviceAddress.split(":")[0];
                String port = serviceAddress.split(":")[1];
    
                //绑定一个端口，返回未来的通道
                ChannelFuture channelFuture = bootstrap.connect(ip, Integer.parseInt(port)).sync();
    
                InvokeRequest invokeRequest = new InvokeRequest();
                invokeRequest.setServiceName(interfaceClass.getName()); //com.bjpowernoede.server.SomeService
                invokeRequest.setMethodName(method.getName());
                invokeRequest.setParamTypes(method.getParameterTypes());
                invokeRequest.setParamValues(args);
    
                //得到一个通道
                Channel channel = channelFuture.channel();
                channel.writeAndFlush(invokeRequest);
    
                //4、当channel被关闭的时候会通知此处关闭chanel（closeFuture方法）
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                eventLoopGroup.shutdownGracefully();
            }
            //客户端接受的结果
            return rpcInvokeHandler.getResult();
        }
    }
    
    
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    /**
     * 客户端（消费者接受结果）
     */
    public class RpcInvokeHandler extends ChannelInboundHandlerAdapter {
        
    
        private Object result;
    
        public Object getResult() {
        
            return result;
        }
    
    
        /**
         * 客户端（消费者接受结果）
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            result = msg;
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        
            super.channelReadComplete(ctx);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            super.exceptionCaught(ctx, cause);
        }
    }

#### **反射类** ####

import java.io.Serializable;
    
    public class InvokeRequest implements Serializable {
        
    
        // com.bjpowernode.service.SomeService
        private  String serviceName;
    
        // doThing(String s, int a, User user)
        private String methodName;
    
        //String s, int a, User user
        private Class<?>[] paramTypes;
    
        //String s, int a, User user
        private Object[] paramValues;
    
        public String getServiceName() {
        
            return serviceName;
        }
    
        public void setServiceName(String serviceName) {
        
            this.serviceName = serviceName;
        }
    
        public String getMethodName() {
        
            return methodName;
        }
    
        public void setMethodName(String methodName) {
        
            this.methodName = methodName;
        }
    
        public Class<?>[] getParamTypes() {
        
            return paramTypes;
        }
    
        public void setParamTypes(Class<?>[] paramTypes) {
        
            this.paramTypes = paramTypes;
        }
    
        public Object[] getParamValues() {
        
            return paramValues;
        }
    
        public void setParamValues(Object[] paramValues) {
        
            this.paramValues = paramValues;
        }
    }
    
    public class Constants {
        
    
        public static final int DEFAULT_PORT = 6666;
    
    }

### 测试 ###

*  启动服务提供者
 *  ![在这里插入图片描述][418cd4dfece1485e99dc1a47c9db667e.png]
 *  启动消费者  
    ![在这里插入图片描述][2bdbc5ec5baa429195f49f00fd658f2b.png]

[3dcc315dfbb14ea1aa6353a799ac2ad5.png]: https://img-blog.csdnimg.cn/3dcc315dfbb14ea1aa6353a799ac2ad5.png
[7f72ff78df3a473ab614049b0545f6e2.png]: https://img-blog.csdnimg.cn/7f72ff78df3a473ab614049b0545f6e2.png
[418cd4dfece1485e99dc1a47c9db667e.png]: https://img-blog.csdnimg.cn/418cd4dfece1485e99dc1a47c9db667e.png
[2bdbc5ec5baa429195f49f00fd658f2b.png]: https://img-blog.csdnimg.cn/2bdbc5ec5baa429195f49f00fd658f2b.png