提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  其它

当前栏目

基于netty实现rpc远程调用

实现 远程 基于 调用 RPC Netty
2023-06-13 09:13:58 时间

文章目录

1.创建API模块

API主要定义对外开放的功能与服务接口

package com.xiepanpan.rpc.api;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20 0020
 * @Description:
 */
public interface IRpcHello {

    String hello(String name);
}
package com.xiepanpan.rpc.api;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20
 * @Description:  计算接口
 */
public interface IRpcCalc {

    /**
     * 加
     * @param a
     * @param b
     * @return
     */
    public int add(int a,int b);

    /**
     * 减
     * @param a
     * @param b
     * @return
     */
    public int sub(int a,int b);

    /**
     * 乘
     * @param a
     * @param b
     * @return
     */
    public int mult(int a,int b);

    /**
     * 除
     * @param a
     * @param b
     * @return
     */
    public int div(int a,int b);
}

##2. 创建自定义协议

Netty中的HTTP处理要Netty内置的HTTP的编解码器来完成解析。现在我们来看自定义协议如何设定。在Netty中完成一个自定义协议其实非常简单，只需要定义一个普通的Java类即可。我们现在手写RPC主要是为了完成对Java代码的远程调用，类似于RMI（Remote Method Invocation，远程方法调用），大家应该都很熟悉了吧。在远程调用Java代码时，哪些内容是必须由网络来传输的呢？譬如，服务名称？需要调用该服务的哪个方法？方法的实参是什么？这些信息都需要通过客户端传送到服务端。

package com.xiepanpan.rpc.core.msg;

import java.io.Serializable;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20
 * @Description: 自定义传输协议的内容
 */
public class InvokerMsg implements Serializable {

    /**
     * 服务名称
     */
    private String className;
    /**
     *  调用哪个方法
     */
    private String methodName;
    /**
     * 参数列表
     */
    private Class<?>[] params;
    /**
     * 参数值
     */
    private Object[] values;

    public String getClassName() {
        return className;
    }

    public void setClassName(String className) {
        this.className = className;
    }

    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    public Class<?>[] getParams() {
        return params;
    }

    public void setParams(Class<?>[] params) {
        this.params = params;
    }

    public Object[] getValues() {
        return values;
    }

    public void setValues(Object[] values) {
        this.values = values;
    }
}

3 实现Provider业务逻辑

package com.xiepanpan.rpc.provider;

import com.xiepanpan.rpc.api.IRpcHello;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20 0020
 * @Description:
 */
public class RpcHello implements IRpcHello {
    public String hello(String name) {
        return "hello,"+name+"!";
    }
}
package com.xiepanpan.rpc.provider;

import com.xiepanpan.rpc.api.IRpcCalc;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20
 * @Description:
 */
public class RpcCalc implements IRpcCalc {
    public int add(int a, int b) {
         return a+b;
    }

    public int sub(int a, int b) {
        return a-b;
    }

    public int mult(int a, int b) {
        return a*b;
    }

    public int div(int a, int b) {
        return a/b;
    }
}

4 完成Registry服务注册

Registry（注册中心）的主要功能就是负责将所有Provider的服务名称和服务引用地址注册到一个容器中，并对外发布。Registry要启动一个对外的服务，很显然应该作为服务端，并提供一个对外可以访问的端口。先启动一个Netty服务，创建RpcRegistry类，具体代码如下。

package com.xiepanpan.rpc.register;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
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.ConcurrentHashMap;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20 0020
 * @Description: 主要保存可用的服务名称和服务地址
 */
public class RpcRegistry {

    public static ConcurrentHashMap<String, Object> registryMap;

    private int port;

    public RpcRegistry(int port) {
        this.port = port;
    }

    public static void main(String[] args) {
        new RpcRegistry(8080).start();
    }

    public void start() {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {

                            ChannelPipeline channelPipeline = ch.pipeline();
                            //拆包粘包处理
                            /**入参有5个，分别解释如下
                             maxFrameLength:框架的最大长度。如果帧的长度大于此值，则将抛出TooLongFrameException
                             lengthFieldoffset:长度属性的偏移量。即对应的长度属性在整个消息数据中的位置
                             lengthFieldLength:长度字段的长度。如果长度属性是int型，那么这个值就是4 (long型就是8)
                             lengthAdjustment:要添加到长度属性值的补偿值
                             initialBytesToStrip:从解码帧中去除的第一个字节数
                             */
                            channelPipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
                            channelPipeline.addLast(new LengthFieldPrepender(4));
                            //编码 解码(JDK 默认序列化)
                            channelPipeline.addLast("encoder",new ObjectEncoder());
                            channelPipeline.addLast("decoder",new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));

                            channelPipeline.addLast(new RegistryHandler());
                        }
                    }).option(ChannelOption.SO_BACKLOG,128)
                    .childOption(ChannelOption.SO_KEEPALIVE,true);
            ChannelFuture channelFuture = serverBootstrap.bind(this.port).sync();

            System.out.println("RPC Registry start listen at"+this.port);
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

在RegistryHandler中实现注册的具体逻辑，上面的代码主要实现服务注册和服务调用的功能。因为所有模块创建在同一个项目中，所以为了简化，服务端没有采用远程调用，而是直接扫描本地Class，然后利用反射调用。代码实现如下。

package com.xiepanpan.rpc.register;

import com.xiepanpan.rpc.core.msg.InvokerMsg;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.io.File;
import java.lang.reflect.Method;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20 0020
 * @Description:
 */
public class RegistryHandler extends ChannelInboundHandlerAdapter {

    public static ConcurrentHashMap<String,Object> registryMap = new ConcurrentHashMap<String,Object>();

    //用来存放class
    private List<String> classCache = new ArrayList<String>();
    
    public RegistryHandler() {
        scanClass("com.xiepanpan.rpc.provider");
        doRegister();
    }

    /**
     * 把扫描到class实例化 放到map中 这就是注册过程
     * 注册的服务名字，叫接口名字
     * 约定优于配置
     */
    private void doRegister() {
        if (classCache.size()==0) {
            return;
        }
        for (String className:classCache) {
            try {
                Class<?> clazz = Class.forName(className);
                //服务名称
                Class<?> interfaces = clazz.getInterfaces()[0];
                registryMap.put(interfaces.getName(),clazz.newInstance());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    /**
     * 扫描出所有的class
     * @param packageName
     */
    private void scanClass(String packageName) {
        URL url = this.getClass().getClassLoader().getResource(packageName.replaceAll("\\.", "/"));
        File dir = new File(url.getFile());
        for(File file: dir.listFiles()) {
            if (file.isDirectory()) {
                scanClass(packageName+"."+file.getName());
            } else {
                classCache.add(packageName+"."+file.getName().replace(".class","").trim());
            }
        }

    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Object result = new Object();
        InvokerMsg request = (InvokerMsg) msg;

        //使用反射调用
        if (registryMap.containsKey(request.getClassName())) {
            Object clazz = registryMap.get(request.getClassName());
            Method method = clazz.getClass().getMethod(request.getMethodName(), request.getParams());
            result = method.invoke(clazz, request.getValues());
        }
        ctx.write(result);
        ctx.flush();
        ctx.close();
    }

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

至此，注册中心的基本功能就完成了。

5 实现Consumer远程调用

梳理一下基本的实现思路，主要完成一个这样的功能：API模块中的接口功能在服务端实现（并没有在客户端实现）。因此，客户端调用API中定义的某一个接口方法时，实际上是要发起一次网络请求去调用服务端的某一个服务。而这个网络请求被注册中心接受，由注册中心先确定需要调用的服务的位置，再将请求转发至真实的服务实现，最终调用服务端代码，将返回值通过网络传输给客户端。整个过程对于客户端而言是完全无感知的，就像调用本地方法一样。具体调用过程如下图所示。

在RpcProxy类的内部实现远程方法调用的代理类，由Netty发送网络请求，具体代码如下。

package com.xiepanpan.rpc.consumer.proxy;

import com.xiepanpan.rpc.core.msg.InvokerMsg;
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.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
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.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.nio.channels.Channel;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20
 * @Description:
 */
public class RpcProxy {
    public static <T> T create(Class<?> clazz) {
        MethodProxy methodProxy = new MethodProxy(clazz);
        Class<?>[] interfaces = clazz.isInterface() ? new Class[]{clazz} : clazz.getInterfaces();
        T result = (T) Proxy.newProxyInstance(clazz.getClassLoader(), interfaces, methodProxy);
        return result;
    }
}

class MethodProxy implements InvocationHandler {

    private Class<?> clazz;
    public MethodProxy(Class<?> clazz) {
        this.clazz = clazz;
    }


    //代理 调用IRpcHello接口中每一个方法的时候，实际就是发起一次网络请求
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //如果传进来的是一个已经实现的具体类 直接忽略
        if (Object.class.equals(method.getDeclaringClass())) {
            try {
                return method.invoke(this,args);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else {
            //如果传进来的是一个接口，我们就远程调用
            return rpcInvoke(method,args);
        }
        return null;
    }

    private Object rpcInvoke(Method method,Object[] args) {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        InvokerMsg msg = new InvokerMsg();

        msg.setClassName(this.clazz.getName());
        msg.setMethodName(method.getName());
        msg.setValues(args);
        msg.setParams(method.getParameterTypes());

        final RpcProxyHandler handler = new RpcProxyHandler();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            //处理拆包粘包问题
                            /**入参有5个，分别解释如下
                             maxFrameLength:框架的最大长度。如果帧的长度大于此值，则将抛出TooLongFrameException
                             lengthFieldoffset:长度属性的偏移量。即对应的长度属性在整个消息数据中的位置
                             lengthFieldLength:长度字段的长度。如果长度属性是int型，那么这个值就是4 (long型就是8)
                             lengthAdjustment:要添加到长度属性值的补偿值
                             initialBytesToStrip:从解码帧中去除的第一个字节数
                             */
                            pipeline.addLast("frameDecoder",new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
                            pipeline.addLast("frameEncoder",new LengthFieldPrepender(4));

                            //处理编解码问题
                            pipeline.addLast("encoder",new ObjectEncoder());
                            pipeline.addLast("decoder",new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));

                            //自己业务的处理
                            pipeline.addLast("handler",handler);
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
            channelFuture.channel().writeAndFlush(msg).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
        return handler.getResult();
    }
}

接收网络调用的返回值，代码如下。

package com.xiepanpan.rpc.consumer.proxy;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20
 * @Description:
 */
public class RpcProxyHandler extends ChannelInboundHandlerAdapter {

    private Object result;
    public Object getResult(){
        return this.result;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        this.result = msg;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("client exception is general");
    }
}

完成客户端调用的代码如下。

package com.xiepanpan.rpc.consumer;

import com.xiepanpan.rpc.api.IRpcCalc;
import com.xiepanpan.rpc.api.IRpcHello;
import com.xiepanpan.rpc.consumer.proxy.RpcProxy;

/**
 * @author: xiepanpan
 * @Date: 2020/8/20 0020
 * @Description:
 */
public class RpcConsumer {

    public static void main(String[] args) {
        IRpcHello rpcHello = RpcProxy.create(IRpcHello.class);
        System.out.println(rpcHello.hello("xp"));

        int a=8,b=2;
        IRpcCalc rpcCalc = RpcProxy.create(IRpcCalc.class);
        System.out.println("a+b="+rpcCalc.add(a,b));
        System.out.println("a-b="+rpcCalc.sub(a,b));
        System.out.println("a*b="+rpcCalc.mult(a,b));
        System.out.println("a/b="+rpcCalc.div(a,b));
    }
}

6. 运行效果

第一步：启动注册中心，运行结果如下图所示。

第二步，运行客户端，调用结果如下图所示。

相关文章