手写一套简单的dubbo（含注册中心）之核心代码

专栏目录:*
Dubbo基础: https://blog.csdn.net/qq_38310244/article/details/125891802
Dubbo实战: https://blog.csdn.net/qq_38310244/article/details/125892120
手写一套简单的dubbo（含注册中心）之编程思想: https://blog.csdn.net/qq_38310244/article/details/125892641
手写一套简单的dubbo（含注册中心）之核心代码: https://blog.csdn.net/qq_38310244/article/details/125892849

上一篇：手写一套简单的dubbo（含注册中心）之编程思想

既然有思路，疑问点都没问题了，就可以开始打码了

代码已上传到GitHub：https://github.com/simple-mine/rpc

创建一个干净的Maven项目，方便依赖的管理

首先，我们需要标记出哪些类需要进行代理，还需要声明项目是否使用我们开发的rpc框架。

EnableRPC

定义一个注解类EnableRPC，标识服务消费者是否启用RPC，并在注解类中定义referencePackage属性，标识哪些类需要生成代理，定义clientRegisterIp声明注册中心的地址

package com.simple.rpc.annotations;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;


@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface EnableRPC {

    //消费者需要注入远程接口的包名
    String referencePackage() default "";

    //注册中心ip
    String clientRegisterIp() default "";
}

StratRPC

定义注解类StratRPC，标识服务提供者是否启用RPC，并在注解类中定义registerPackage属性，标识哪些类是接口的实现类（需要生成代理的类），serverPort属性标识服务监听的端口（服务启动后，需要一个端口监听来自服务消费者的请求），serverRegisterIp声明注册中心地址

package com.simple.rpc.annotations;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;


@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface StratRPC {

    //需要注册的服务所在的包
    String registerPackage() default "";

    //服务监听的端口
    int serverPort() default 0;

    //注册中心ip
    String serverRegisterIp() default "";


}

RegisterService

定义RegisterService注解类，用于标识服务提供者哪些类是真真切切需要生成代理对象的，虽然前面已经需要生成代理类的包，但是不一定该包路径下的类都需要生成代理对象，这是对需要生成代理对象的类更加细化

package com.simple.rpc.annotations;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;


@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface RegisterService {

    String name() default "";
}

Reference

定义注解类Reference，用于标识服务消费者哪些类里需要引用到远程服务提供者提供的服务，在Springboot启动时，为这些成员变量注入代理对象

package com.simple.rpc.annotations;

import java.lang.annotation.*;


@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
public @interface Reference {
    String name() default "";
}

AbstractInterfaceFactory

创建服务消费者创建代理对象工厂抽象类AbstractInterfaceFactory，该类用于创建服务消费者需要生成代理对象的成员变量。该类定义了一个调用远程服务的抽象方法，并定义内部类默认实现该抽象方法的具体实现方式。

package com.simple.rpc.factory;



import com.simple.rpc.socket.SocketRequest;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.logging.Logger;


public abstract class AbstractInterfaceFactory {
    protected static final Logger logger = Logger.getLogger(AbstractInterfaceFactory.class.getName());

    public static AbstractInterfaceFactory getInstance() {
        return new DefaultInterfaceFactory();
    }

    protected AbstractInterfaceFactory(){}

    public  T getWebService(Class oldInterface) {
        InterfaceHandler intr = new InterfaceHandler(this);
        return  (T) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{oldInterface}, new InterfaceHandler(this));
    }

    
    protected abstract Object remoteCall(Method methodName, Object[] args);

    
    private static final class InterfaceHandler implements InvocationHandler {
        private AbstractInterfaceFactory factory;

        public InterfaceHandler(AbstractInterfaceFactory factory) {
            this.factory = factory;
        }

        @Override
        public Object invoke(Object proxy, Method method, Object[] args) {
            String remoteMethodName = method.getName();
            logger.info("开始调用接口：" + remoteMethodName);
            Object rst = factory.remoteCall(method, args);
            logger.info("完成调用");
            return rst;
        }
    }

    
    private static final class DefaultInterfaceFactory extends AbstractInterfaceFactory {
        @Override
        protected Object remoteCall(Method method, Object[] args) {
            logger.info("远程方法调用中.....");
            return SocketRequest.sendRequest(method,args);
        }
    }
}

ProxyConfig

创建ProxyConfig类，用于扫描服务消费者哪些类需要注入或实例化代理对象

package com.simple.rpc.config;

import com.simple.rpc.annotations.Reference;
import com.simple.rpc.factory.AbstractInterfaceFactory;
import org.reflections.Reflections;
import org.reflections.scanners.FieldAnnotationsScanner;
import org.reflections.util.ClasspathHelper;
import org.reflections.util.ConfigurationBuilder;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.util.Set;


@Component
public class ProxyConfig{

    public static void initReference(ApplicationContext applicationContext,String referencePackage){
        //反射工具包，指明扫描路径
        Reflections reflections = new Reflections(new ConfigurationBuilder()
                .setUrls(ClasspathHelper.forPackage(referencePackage))
                .setScanners(new FieldAnnotationsScanner()));

        Set fields = reflections.getFieldsAnnotatedWith(Reference.class);
        for (Field field : fields) {
            //获取需要远程调用的属性字段
            Reference reference = field.getAnnotation(Reference.class);
            if (reference != null) {
                try {
                    //获取代理对象
                    Object webService = AbstractInterfaceFactory.getInstance().getWebService(field.getType());
                    //允许字段赋值
                    field.setAccessible(true);
                    Object bean;
                    if (applicationContext == null){
                        //非容器化启动时
                        bean = field.getDeclaringClass().newInstance();
                    }else {
                        //获取容器里该字段的类
                        bean = applicationContext.getBean(field.getDeclaringClass());
                    }
                    //对字段进行赋值
                    field.set(bean, webService);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

ServerInterfaceFactory

创建ServerInterfaceFactory类，用于为服务提供者创建代理对象

package com.simple.rpc.factory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;


public class ServerInterfaceFactory implements InvocationHandler {
    
    private final Object target;

    public ServerInterfaceFactory(Object target) {
        this.target = target;
    }

    
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //调用目标方法
        Object result = null;
        try {
            //前置通知
            result = method.invoke(target, args);
        } catch (Exception e) {
            e.printStackTrace();
            //异常通知, 可以访问到方法出现的异常
            System.out.println( "方法调用出现了异常");
        }
        //后置通知.
        System.out.println( "方法调用完成！");
        return result;
    }

    
    public Object getProxy() {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),target.getClass().getInterfaces(),this);
    }
}

SocketRequest

创建SocketRequest类，用于实现服务消费者调用远程服务提供者的具体实现（通过Socket方式进行调用）

package com.simple.rpc.socket;

import com.simple.rpc.main.RPCMain;
import com.simple.util.entiy.TCPConfig;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.Socket;


public class SocketRequest {

    public static Object sendRequest(Method method, Object[] args){
        // 获取到远程类
        String methodName = method.getName();
        String serverName = method.getDeclaringClass().getTypeName() + "."+methodName;
        TCPConfig tcpConfig = RPCMain.getTcpMap().get(serverName);
        if (tcpConfig == null){
            throw new RuntimeException("can not find provider");
        }
        return sendAndReceive(tcpConfig.getIp(),tcpConfig.getPort(),serverName,methodName,args);
    }

    private static Object sendAndReceive(String ip, int port,String serverName,String methodName, Object[] args){
        //开启一个链接，需要指定地址和端口
        try (Socket client = new Socket(ip, port)){

            //向输出流中写入数据，传向服务端
            ObjectOutputStream objectOutputStream = new ObjectOutputStream(client.getOutputStream());
            objectOutputStream.writeObject(serverName);
            objectOutputStream.writeObject(methodName);
            objectOutputStream.writeObject(args);

            //从输入流中解析数据，输入流来自服务端的响应
            ObjectInputStream objectInputStream = new ObjectInputStream(client.getInputStream());
            return objectInputStream.readObject();
        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException("invoke error");
        }
    }
}

Execute

创建Execute类，该类创建一个线程监听对外暴露的端口，实现处理远程消费者调用方法时，通过代理对象找到服务提供者对应的实现方法，并执行后返回给服务消费者

package com.simple.rpc.main;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.simple.rpc.factory.ServerInterfaceFactory;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class Execute {

    private static ThreadFactory namedFactory = new ThreadFactoryBuilder().setNameFormat("provid-thread-%d").build();

    private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new
            LinkedTransferQueue<>(), namedFactory, new ThreadPoolExecutor.AbortPolicy());

    public static void startTpcServer(String ip, int serverPort){
        threadPool.execute(()->{
            executeRequest(ip,serverPort);
        });
    }

    private static void executeRequest(String ip, int serverPort){
        try {
            //封装服务端地址
            InetAddress serverAddress = InetAddress.getByName(ip);
            //建立服务端
            try (ServerSocket service = new ServerSocket(serverPort, 10, serverAddress)) {
                while (true) {
                    //接受一个连接，该方法会阻塞程序，直到一个链接到来
                    try (Socket connect = service.accept()) {

                        //解析输入流，该输入流来自客户端
                        ObjectInputStream objectInputStream = new ObjectInputStream(connect.getInputStream());

                        String serverName = (String) objectInputStream.readObject();
                        String methodName = (String) objectInputStream.readObject();
                        Object[] args = (Object[]) objectInputStream.readObject();

                        if (serverName != null && methodName != null && args != null){
                            Map> serviceMap = RPCMain.getServiceMap();
                            Class aClass1 = serviceMap.get(serverName);

                            Object invoke = "";
                            //查找对应的方法
                            Method[] methods = aClass1.getMethods();
                            for (Method method : methods) {
                                //调用方法
                                if (methodName.equals(method.getName())){
                                    //创建代理对象
                                    invoke  = new ServerInterfaceFactory(aClass1.newInstance()).invoke(aClass1, method, args);
                                }
                            }
                            //组建响应信息
                            OutputStream out = connect.getOutputStream();
                            ObjectOutputStream objectOutputStream = new ObjectOutputStream(out);
                            objectOutputStream.writeObject(invoke);
                        }else {
                            //获取输入流，并通过向输出流写数据的方式发送响应
                            OutputStream out = connect.getOutputStream();
                            ObjectOutputStream objectOutputStream = new ObjectOutputStream(out);
                            objectOutputStream.writeObject("can not find provid");
                        }

                    } catch (Throwable e) {
                        e.printStackTrace();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
    }
}

ClientServer

创建ClientServer类，该类使用netty，与服务端建立连接（此处默认使用9999端口，也可以通过注解配置的方式），并在启动时将自身对外提供的服务（不管有没有）发送给服务端。

package com.simple.rpc.register;

import com.simple.util.entiy.TCPConfig;
import com.simple.util.serializer.DecoderSerializer;
import com.simple.util.serializer.EncoderSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.Map;


public class ClientServer {

    private final static Integer NETTYPORT = 9999;

    public static void run(String ip,Map tcpConfigMap) {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            //创建bootstrap对象配置参数
            Bootstrap bootstrap = new Bootstrap();
            //设置线程组
            bootstrap.group(eventLoopGroup)
                    //设置客户端通道实现类型
                    .channel(NioSocketChannel.class)
                    //初始化通道
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ch.pipeline().addLast(new EncoderSerializer());
                            ch.pipeline().addLast(new DecoderSerializer());
                            ch.pipeline().addLast(new ClientServerHandler());
                        }
                    });
            System.out.println("客户端准备完毕");
            //连接服务器
            Channel channel = bootstrap.connect(ip, NETTYPORT).sync().channel();
            channel.writeAndFlush(tcpConfigMap);
            channel.flush();
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //关闭线程组
            eventLoopGroup.shutdownGracefully();
        }
    }
}

ClientServerHandler

创建ClientServerHandler类，用于接收netty服务端返回的消息。当启动服务服务之后，服务端会返回服务端所保有的所有服务列表，本地将服务列表保存到内存，在需要时直接查询对应的服务进行调用。

package com.simple.rpc.register;

import com.simple.rpc.main.RPCMain;
import com.simple.util.entiy.TCPConfig;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Map;


@ChannelHandler.Sharable
public class ClientServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Map map = (Map) msg;
        Map tcpMap = RPCMain.getTcpMap();
        //向本地的tcpMap中添加注册中心返回的其他服务信息
        for (Object key : map.keySet()) {
            tcpMap.put(key.toString(), (TCPConfig) map.get(key));
        }
        RPCMain.setTcpMap(tcpMap);
    }
}

RPCMain

创建RPCMain类，该类是整个RPC启动时所执行的流程

package com.simple.rpc.main;

import com.simple.rpc.annotations.EnableRPC;
import com.simple.rpc.annotations.RegisterService;
import com.simple.rpc.annotations.StratRPC;
import com.simple.rpc.config.ProxyConfig;
import com.simple.rpc.register.ClientServer;
import com.simple.util.entiy.TCPConfig;
import org.reflections.Reflections;
import org.reflections.scanners.FieldAnnotationsScanner;
import org.reflections.scanners.SubTypesScanner;
import org.reflections.scanners.TypeAnnotationsScanner;
import org.reflections.util.ClasspathHelper;
import org.reflections.util.ConfigurationBuilder;
import org.springframework.beans.BeansException;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;


@Component
@Order(1)
public class RPCMain implements CommandLineRunner, ApplicationContextAware {

    private static final Logger logger = Logger.getLogger(RPCMain.class.getName());

    private  static String registerPackage;

    private  static String hostAddress;

    private  static Integer serverPort;

    private  static String registerIp = null;

    private  static Map> serviceMap = new HashMap<>();

    private  static Map tcpMap = new HashMap<>();

    private  static ApplicationContext context;

    private static String getMainClassName() {
        //获取到启动类
        StackTraceElement[] stackTraceElements = new RuntimeException().getStackTrace();
        for (StackTraceElement stackTraceElement : stackTraceElements) {
            if ("main".equals(stackTraceElement.getMethodName())) {
                return stackTraceElement.getClassName();
            }
        }
        return "";
    }

    private  void initTcpServerClass(){
        //只扫描指定的包路径
        Reflections reflections = new Reflections(new ConfigurationBuilder()
                .setUrls(ClasspathHelper.forPackage(registerPackage))
                .setScanners(new FieldAnnotationsScanner(),new TypeAnnotationsScanner(),new SubTypesScanner(false)));
        Set> serverClass = reflections.getTypesAnnotatedWith(RegisterService.class);
        for (Class aClass : serverClass) {
            RegisterService service = aClass.getAnnotation(RegisterService.class);
            if (service == null) {
                continue;
            }
            Class[] interfaces = aClass.getInterfaces();
            for (Class anInterface : interfaces) {
                Method[] methods = anInterface.getMethods();
                for (Method method : methods) {
                    String name = anInterface.getCanonicalName()+"."+ method.getName();
                    serviceMap.put(name, aClass);
                    TCPConfig tcpConfig = new TCPConfig(name,hostAddress,serverPort,Boolean.TRUE);
                    tcpMap.put(name,tcpConfig);
                }
            }
        }
    }

    private  void getHost() {
        try{
            InetAddress addr = InetAddress.getLocalHost();
            hostAddress = addr.getHostAddress();
        }catch (UnknownHostException e){
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }
    }

    public static Map getTcpMap() {
        return tcpMap;
    }

    public static void setTcpMap(Map tcpMap) {
        RPCMain.tcpMap = tcpMap;
    }

    public static Map> getServiceMap() {
        return serviceMap;
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        context = applicationContext;
    }


    @Override
    public void run(String... args) throws Exception {
        String mainClassName = getMainClassName();
        Class aClass = null;
        try {
            aClass = ClassLoader.getSystemClassLoader().loadClass(mainClassName);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

        if (aClass == null){
            throw new RuntimeException("can not find main class");
        }

        EnableRPC enabletcp = aClass.getAnnotation(EnableRPC.class);
        if (enabletcp != null){
            String referencePackage = enabletcp.referencePackage();
            if ("".equals(referencePackage.replaceAll("\s*|t|r|n", ""))){
                throw new RuntimeException("@com.simple.tcp.annotations.EnableTCP.referencePackage can not be empty");
            }
            String clientRegisterIp = enabletcp.clientRegisterIp();
            if ("".equals(clientRegisterIp.replaceAll("\s*|t|r|n", ""))){
                throw new RuntimeException("@com.simple.tcp.annotations.StratTCP.clientRegisterIp can not be empty");
            }
            registerIp = clientRegisterIp;
            //为service对象注入代理
            ProxyConfig.initReference(context,referencePackage);
        }

        StratRPC annotation = aClass.getAnnotation(StratRPC.class);
        //根据启动类上是否存在注解判断是否启动tcp
        if (annotation != null){
            registerPackage = annotation.registerPackage();
            if ("".equals(registerPackage.replaceAll("\s*|t|r|n", ""))){
                throw new RuntimeException("@com.simple.tcp.annotations.StratTCP.registerPackage can not be empty");
            }

            String serverRegisterIp = annotation.serverRegisterIp();
            if ("".equals(serverRegisterIp.replaceAll("\s*|t|r|n", ""))){
                throw new RuntimeException("@com.simple.tcp.annotations.StratTCP.serverRegisterIp can not be empty");
            }

            serverPort = annotation.serverPort();
            if (serverPort == 0){
                throw new RuntimeException("@com.simple.tcp.annotations.StratTCP.serverPort can not be empty");
            }
            if (serverPort == 9999){
                throw new RuntimeException("Regiter center has use this port.Please use other port");
            }

            getHost();

            //初始化服务类
            initTcpServerClass();



            //启动线程监听
            Execute.startTpcServer(hostAddress,serverPort);

            if (registerIp != null && serverRegisterIp.equals(registerIp)){
                throw new RuntimeException("@com.simple.tcp.annotations.StratTCP.serverRegisterIp must equals @com.simple.tcp.annotations.EnableTCP.clientRegisterIp");
            }
            registerIp = serverRegisterIp;
        }

        // 启动netty,发布/订阅服务
        ClientServer.run(registerIp,tcpMap);
    }
}

我们知道，Springboot是不会将第三方jar包里的配置信息在启动时自动进行加载，要么在代码里显示的将第三方jar包的配置注入到容器，或者使用spring.factories文件。为了尽量达到无侵入，这里使用spring.factories

在resource文件夹下新建META-INF文件夹，在该文件夹下新建spring.factories指定核心启动类的包路径

org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.simple.rpc.main.RPCMain

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EjxzPkNW-1658301035643)(E:studydubboimage微信截图_20220719170921.png)]

由于netty传输数据时需要经过编码（这个是坑），这里创建三个处理netty传输服务列表信息的编码与解密类。netty客户端和netty服务端都必须用同一套编码工具类

DecoderSerializer

解码

package com.simple.util.serializer;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;


public class DecoderSerializer extends ByteToMessageDecoder {
    private static final int LENGTH = 4;

    
    @Override
    protected void decode(ChannelHandlerContext arg0, ByteBuf in, List