关于程序员:用Netty-自己实现dubbo-RPC

用Netty 本人实现dubbo RPC

RPC 的根本介绍

RPC (Remote Procedure Call) 近程过程调用，是一个计算机通信协议。该协定容许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额定的为这个交互编程。也就是说能够达到两个或者多个应用程序部署在不同的服务器上，他们之间的调用都像是本地办法调用一样。RPC 的调用如下图。

罕用的RPC 框架有阿里的dubbo，Google的gRPC，Go 语言的rpcx，Apache的thrift，Spring的Spring Cloud.

RPC 调用的过程

在RPC 中，Client 端叫做服务消费者，Server 叫做服务提供者。

调用流程阐明

服务生产方（client）以本地调用形式调用服务
client stub 接管到调用后负责将办法、参数等封装成可能进行网络传输的音讯体
client stub 将音讯进行编码并发送到服务端
server stub 接管到音讯后进行解码
server stub 依据解码后果调用本地的服务
本地服务执行并将后果返回给server stub
server stub 将返回导入后果进行编码并发送给生产方
client stub 接管到音讯并进行解码
服务生产方（client）失去后果

其中，RPC 框架的指标就是把2-8 这些步骤封装起来，用户无需关怀这些细节，能够像调用本地办法一样即可实现近程服务调用。

本人实现dubbo RPC

需要阐明

dubbo 底层应用了Netty 作为网络通信框架，要求用netty 实现一个简略的RPC框架。
模拟dubbo，消费者和提供者约定接口和协定，消费者近程调用提供者的服务，提供者返回一个字符串，消费者打印提供者返回的数据。底层网络通信给予Netty 4.x

设计说明

创立一个接口，定义形象办法。用于消费者和提供者之间的约定。
创立一个提供者，该类须要监听消费者的申请，并依照约定返回数据。
创立一个消费者，该类须要通明的调用本人不存在的办法，外部须要应用netty申请提供者返回数据
开发的剖析图如下:

代码实现

1.定义对立的接口

//这个是接口，是服务提供方和 服务生产方都须要
public interface HelloService {

    String hello(String mes);
}

2.服务的提供方

// 先写一个实现刚刚接口的办法
public class HelloServiceImpl implements HelloService {

    private static int count = 0;
    //当有生产方调用该办法时， 就返回一个后果
    @Override
    public String hello(String mes) {
        System.out.println("收到客户端音讯=" + mes);
        //依据mes 返回不同的后果
        if(mes != null) {
            return "你好客户端, 我曾经收到你的音讯 [" + mes + "] 第" + (++count) + " 次";
        } else {
            return "你好客户端, 我曾经收到你的音讯 ";
        }
    }
}

//而后紧接着写基于netty的服务端的解决handler以及nettyServer   这里服务器这边handler比较简单
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //获取客户端发送的音讯，并调用服务
        System.out.println("msg=" + msg);
        //客户端在调用服务器的api 时，咱们须要定义一个协定
        //比方咱们要求 每次发消息是都必须以某个字符串结尾 "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) throws Exception {
        ctx.close();
    }
}

public class NettyServer {


    public static void startServer(String hostName, int port) {
        startServer0(hostName,port);
    }

    //编写一个办法，实现对NettyServer的初始化和启动

    private static void startServer0(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) throws Exception {
                                          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();
        }

    }
}

// 最初写一个服务端的启动类
//ServerBootstrap 会启动一个服务提供者，就是 NettyServer
public class ServerBootstrap {
    public static void main(String[] args) {

        //代码代填..
        NettyServer.startServer("127.0.0.1", 7000);
    }
}

3.服务的生产方

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("(proxy, method, args) 进入...." + (++count) + " 次");
                    //{}  局部的代码，客户端每调用一次 hello, 就会进入到该代码
                    if (client == null) {
                        initClient();
                    }

                    //设置要发给服务器端的信息
                    //providerName 协定头 args[0] 就是客户端调用api hello(???), 参数
                    client.setPara(providerName + args[0]);

                    //
                    return executor.submit(client).get();

                });
    }

    //初始化客户端
    private static void initClient() {
        client = new NettyClientHandler();
        //创立EventLoopGroup
        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) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                pipeline.addLast(new StringDecoder());
                                pipeline.addLast(new StringEncoder());
                                pipeline.addLast(client);
                            }
                        }
                );

        try {
            bootstrap.connect("127.0.0.1", 7000).sync();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {

    private ChannelHandlerContext context;//上下文
    private String result; //返回的后果
    private String para; //客户端调用办法时，传入的参数


    //与服务器的连贯创立后，就会被调用, 这个办法是第一个被调用(1)
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(" channelActive 被调用  ");
        context = ctx; //因为咱们在其它办法会应用到 ctx
    }

    //收到服务器的数据后，调用办法 (4)
    //
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(" channelRead 被调用  ");
        result = msg.toString();
        notify(); //唤醒期待的线程
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

    //被代理对象调用, 发送数据给服务器，-> wait -> 期待被唤醒(channelRead) -> 返回后果 (3)-》5
    @Override
    public synchronized Object call() throws Exception {
        System.out.println(" call1 被调用  ");
        context.writeAndFlush(para);
        //进行wait
        wait(); //期待channelRead 办法获取到服务器的后果后，唤醒
        System.out.println(" call2 被调用  ");
        return  result; //服务方返回的后果

    }
    //(2)
    void setPara(String para) {
        System.out.println(" setPara  ");
        this.para = para;
    }
}

// client 端的启动类
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);

        for (;; ) {
            Thread.sleep(2 * 1000);
            //通过代理对象调用服务提供者的办法(服务)
            String res = service.hello("你好 dubbo~");
            System.out.println("调用的后果 res= " + res);
        }
    }
}

关于程序员:用Netty-自己实现dubbo-RPC

用Netty 本人实现dubbo RPC

RPC 的根本介绍

RPC 调用的过程

调用流程阐明

本人实现dubbo RPC

需要阐明

设计说明

代码实现

1.定义对立的接口

2.服务的提供方

3.服务的生产方

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

关于程序员:用Netty-自己实现dubbo-RPC

用Netty 本人实现dubbo RPC

RPC 的根本介绍

RPC 调用的过程

调用流程阐明

本人实现dubbo RPC

需要阐明

设计说明

代码实现

1.定义对立的接口

2.服务的提供方

3.服务的生产方

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复