从零学Netty（十一）Netty实现RPC

红太狼 2022-12-17 03:28 170阅读 0赞

# 简介 #

远程过程调用，简单的理解是一个节点请求另一个节点提供的服务

简单理解 RPC就是要像调用本地的函数一样去调远程函数

原理图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70][]

# 实例demo #

### 通用 ###

通用接口  用于生产端实现具体业务  与  消费点进行消费

/**
     * 通用接口
     *
     * @author LionLi
     */
    public interface HelloService {
        String hello(String mes);
    }

### 生产端 ###

实现通用接口

/**
     * 接口实现
     *
     * @author LionLi
     */
    public class HelloServiceImpl implements HelloService {
    
        private static int count = 0;
    
        @Override
        public String hello(String mes) {
            System.out.println("收到客户端消息 => " + mes);
            return "服务端 => 收到消息 [" + mes + "] 第" + (++count) + " 次";
        }
    }

生产端启动类  用于启动netty服务器

/**
     * 生产端启动类
     *
     * @author LionLi
     */
    public class ProviderBootstrap {
    
        public static void main(String[] args) {
            NettyServer.startServer("localhost", 8088);
        }
    
    }

### 消费端 ###

对通用接口进行消费调用

/**
     * 消费端启动类
     *
     * @author LionLi
     */
    public class ClientBootstrap {
        // 协议头
        public static final String providerName = "HelloService#hello#";
    
        public static void main(String[] args) throws  Exception{
            NettyClient customer = new NettyClient();
            // 创建代理对象
            HelloService service = (HelloService) customer.getBean(HelloService.class, providerName);
            while(true) {
                Thread.sleep(2 * 1000);
                // 通过代理对象调用服务提供者的方法
                String res = service.hello("你好服务端");
                System.out.println("结果 => " + res);
            }
        }
    }

### 服务端 ###

使用netty构建服务器  用于接收和处理请求

/**
     * 服务端
     *
     * @author LionLi
     */
    public class NettyServer {
    
        public static void startServer(String hostname, int port) {
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                                          @Override
                                          protected void initChannel(SocketChannel ch) {
                                              ChannelPipeline pipeline = ch.pipeline();
                                              pipeline.addLast(new StringDecoder());
                                              pipeline.addLast(new StringEncoder());
                                              pipeline.addLast(new NettyServerHandler());
                                          }
                                      }
                        );
                ChannelFuture channelFuture = serverBootstrap.bind(hostname, port).sync();
                System.out.println("服务端正在监听......");
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
    
        }
    }

服务端业务处理

/**
     * 服务端业务处理
     *
     * @author LionLi
     */
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            System.out.println("msg => " + msg);
            // 定义一个协议,比如: "HelloService#hello#你好"
            if(msg.toString().startsWith(ClientBootstrap.providerName)) {
                // 截取协议后的内容
                String result = new HelloServiceImpl().hello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
                ctx.writeAndFlush(result);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            ctx.close();
        }
    }

### 客户端 ###

用于创建和发送请求  使用代理模式 来创建请求类

/**
     * 客户端
     *
     * @author LionLi
     */
    public class NettyClient {
    
        private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        private static NettyClientHandler client;
        private int count = 0;
    
        /**
         * 代理模式，获取一个代理对象
         */
        public Object getBean(final Class<?> serivceClass, final String providerName) {
            return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                    new Class<?>[]{serivceClass}, (proxy, method, args) -> {
                        System.out.println("进入代理" + (++count) + " 次");
                        if (client == null) {
                            initClient();
                        }
                        // 设置要发给服务器端的信息
                        // providerName 协议头
                        // args[0] 是客户端传递的参数
                        client.setParam(providerName + args[0]);
                        return executor.submit(client).get();
                    });
        }
    
        /**
         * 初始化客户端
         */
        private static void initClient() {
            client = new NettyClientHandler();
            NioEventLoopGroup group = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                                 @Override
                                 protected void initChannel(SocketChannel ch) {
                                     ChannelPipeline pipeline = ch.pipeline();
                                     pipeline.addLast(new StringDecoder());
                                     pipeline.addLast(new StringEncoder());
                                     pipeline.addLast(client);
                                 }
                             }
                    );
            try {
                bootstrap.connect("localhost", 8088).sync();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

业务处理

/**
     * 客户端业务处理
     *
     * @author LionLi
     */
    public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {
    
        private ChannelHandlerContext context;
        private String result;
        private String param;
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            System.out.println("初始化连接......");
            context = ctx;
        }
    
        @Override
        public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
            System.out.println("读取数据......");
            result = msg.toString();
            //唤醒等待的线程
            notify();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            ctx.close();
        }
    
        /**
         * 被代理对象调用, 发送数据给服务器
         */
        @Override
        public synchronized Object call() throws Exception {
            System.out.println("call 进入");
            context.writeAndFlush(param);
            //等待 channelRead 方法获取到服务器的结果后，唤醒
            wait();
            System.out.println("call 返回");
            return  result;
    
        }
    
        public void setParam(String param) {
            this.param = param;
        }
    }

# 测试 #

首先启动生产端

![20201021173119810.png][]

启动消费端 查看日志

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 1][]

客户端发送了三次请求  服务端响应了三次

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 2][]

项目已上传到gitee

地址: [netty-demo][]

如果帮到您了,请帮忙点个star

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70]: /images/20221123/572a3ad373754d2cbba2b919abb7f783.png
[20201021173119810.png]: /images/20221123/a41179253c044da999713a377cbcaf39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 1]: /images/20221123/8706a567bfb646c2972a96ac1e1c2a7e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ2MTI4MQ_size_16_color_FFFFFF_t_70 2]: /images/20221123/16225b11783f495eb8b6106b70ab663d.png
[netty-demo]: https://gitee.com/JavaLionLi/LionLi-Labs/tree/master/netty-demo