【微服务】Java模拟实现dubbo框架

前言

在对dubbo有了较为深入的使用和理解后，来尝试从dubbo框架的角度重新认识下它，对照着dubbo官方的这张图进行反复的理解后，我们可以从已有掌握的技术出发，来尝试编写一个简单的dubbo实现。

dubbo技术实现

dubbo详细的说明这里就不再一一详述了，重点理解下面这张图

从这张图，可以得出下面几点能够指导我们编码的要点：

服务提供方和服务消费端为两个JVM进程；
服务提供方需要将服务注册到某个地方（注册中心），方便消费方找到服务并调用；
服务消费方从注册中心找到服务后，像调用本地接口一样调用注册中心的服务；

上面三点的补充说明

1、服务提供方和服务消费端为两个JVM进程

这意味着服务提供者和消费者的代码需要分开

2、服务提供方需要将服务注册到某个地方

需要提供一种方式，可以将服务接口注册上去，并被服务消费方的JVM获取到

3、消费方像调用本地接口一样调用注册中心的服务

意味着需要通过某种机制，能够将服务接口的代理实现进行加载，在上图中注意有个关键字 invoke

基于上面的实现思路，接下来让我们编写代码进行实现吧

一、创建maven工程

导入基本的依赖

<dependencies><dependency><groupId>commons-httpclient</groupId><artifactId>commons-httpclient</artifactId><version>3.1</version></dependency><dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient</artifactId><version>4.5.6</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>2.3.2.RELEASE</version></dependency><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.16.Final</version></dependency><dependency><groupId>org.apache.tomcat.embed</groupId><artifactId>tomcat-embed-core</artifactId><version>9.0.12</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-io</artifactId><version>1.3.2</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.4</version><scope>provided</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.51</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>5.0.0</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-x-discovery</artifactId><version>5.0.0</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.5.8</version></dependency></dependencies>

二、服务提供方编码实现

按照上面的编写思路，服务提供方要实现的功能点主要包括如下几点

提供服务接口实现；
作为一个单独的JVM进程，可以考虑 jetty，tomcat，或者netty等；
能够解析服务消费方传递过来的请求，并找到服务接口的实现，并返回处理结果；

1、服务接口

public interface HelloService {String sayHello(String message);}

2、接口实现

public class HelloServiceImpl implements HelloService {@Overridepublic String sayHello(String message) {return "hello : " + message;}
}

3、使用内嵌式的tomcat容器发布服务

提供一个HttpServer

import org.apache.catalina.*;
import org.apache.catalina.connector.Connector;
import org.apache.catalina.core.StandardContext;
import org.apache.catalina.core.StandardEngine;
import org.apache.catalina.core.StandardHost;
import org.apache.catalina.startup.Tomcat;public class HttpServer {public void start(String hostname, Integer port) {Tomcat tomcat = new Tomcat();Server server = tomcat.getServer();Service service = server.findService("Tomcat");Connector connector = new Connector();connector.setPort(port);Engine engine = new StandardEngine();engine.setDefaultHost(hostname);Host host = new StandardHost();host.setName(hostname);String contextPath = "";Context context = new StandardContext();context.setPath(contextPath);context.addLifecycleListener(new Tomcat.FixContextListener());host.addChild(context);engine.addChild(host);service.setContainer(engine);service.addConnector(connector);//将 Tomcat 接收到的所有请求都交给自定义的 DispatcherServlet 来处理tomcat.addServlet(contextPath, "dispatcher", new MyServlet());context.addServletMappingDecoded("/*", "dispatcher");try {tomcat.start();tomcat.getServer().await();} catch (LifecycleException e) {e.printStackTrace();}}}

4、自定义的Servlet

使用自定义的Servlet ，可以让程序得到一个的扩展

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;public class MyServlet extends HttpServlet {@Overrideprotected void service(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {new HttpServletHandler().handler(req,resp);}
}

5、自定义HttpServletHandler

该类用于处理来自服务消费方的请求，并返回结果

import com.congge.framework.Invocation;
import com.congge.framework.resgister.LocalRegister;
import org.apache.commons.io.IOUtils;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;public class HttpServletHandler {public void handler(HttpServletRequest req, HttpServletResponse resp) {try {Invocation invocation = (Invocation) new ObjectInputStream(req.getInputStream()).readObject();String interfaceName = invocation.getInterfaceName();Class implClass = LocalRegister.get(interfaceName);try {Method method = implClass.getMethod(invocation.getMethodName(), invocation.getParamTypes());try {String result = (String)method.invoke(implClass.newInstance(), invocation.getParams());System.out.println("执行的结果:" + result);IOUtils.write(result,resp.getOutputStream());} catch (IllegalAccessException e) {e.printStackTrace();} catch (InvocationTargetException e) {e.printStackTrace();} catch (InstantiationException e) {e.printStackTrace();}} catch (NoSuchMethodException e) {e.printStackTrace();}} catch (IOException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();}}}

6、服务注册中心

尽管是一个简单的实现，发布出去的服务接口也需要一个注册中心承载，这里我们先使用一个简单的map结构实现服务注册中心，自定义LocalRegister，简单来说，map的key为接口名，value为实现类

import java.util.HashMap;
import java.util.Map;public class LocalRegister {private static Map<String,Class> map = new HashMap<>();public static void register(String interfaceName,Class implClass){map.put(interfaceName,implClass);}public static Class get(String interfaceName){return map.get(interfaceName);}}

6、服务提供方启动类

public class Provider {public static void main(String[] args) {//注册接口LocalRegister.register(HelloService.class.getName(),HelloServiceImpl.class);HttpServer httpServer = new HttpServer();httpServer.start("localhost",8081);}}

启动main程序，这样服务生产方的接口就发布出去了，暴露的端口为8081

三、服务消费方编码实现

照上面的编写思路，服务消费方要实现的功能点主要包括如下几点

从注册中心获取特定的服务接口并执行调用；
作为一个单独的JVM进程，可以考虑 jetty，tomcat，或者netty等；
需要有个类，用于集中处理发起接口请求调用

1、提供一个HttpClient

该类用于发起请求，请求特定的服务接口，试想在真实的调用中，比如是一个springboot的应用，服务消费方要调用生产者的服务，也是要走http协议或者通过netty的方式进行调用的；

import com.congge.framework.Invocation;
import org.apache.commons.io.IOUtils;import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;public class HttpClient {public String send(String hostname, Integer port, Invocation invocation) {String result = null;try {URL url = new URL("http", hostname, port, "/");try {HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();httpURLConnection.setRequestMethod("POST");httpURLConnection.setDoOutput(true);OutputStream outputStream = httpURLConnection.getOutputStream();ObjectOutputStream oos = new ObjectOutputStream(outputStream);oos.writeObject(invocation);oos.flush();InputStream inputStream = httpURLConnection.getInputStream();result = IOUtils.toString(inputStream);return result;} catch (IOException e) {e.printStackTrace();}} catch (MalformedURLException e) {e.printStackTrace();}return result;}
}

2、提供一个用于封装请求参数的对象类

该类封装了从服务消费方发出请求的完整参数对象，比如请求的接口名，方法名，参数类型等

public class Invocation implements Serializable {private String interfaceName;  //接口名private String methodName;   //方法名private Class[] paramTypes;  //方法参数类型列表private Object[] params;     //方法参数值列表public Invocation(String interfaceName, String methodName, Class[] paramTypes, Object[] params) {this.interfaceName = interfaceName;this.methodName = methodName;this.paramTypes = paramTypes;this.params = params;}public String getInterfaceName() {return interfaceName;}public void setInterfaceName(String interfaceName) {this.interfaceName = interfaceName;}public String getMethodName() {return methodName;}public void setMethodName(String methodName) {this.methodName = methodName;}public Object[] getParams() {return params;}public void setParams(Object[] params) {this.params = params;}public Class[] getParamTypes() {return paramTypes;}public void setParamTypes(Class[] paramTypes) {this.paramTypes = paramTypes;}
}

3、消费方启动类

import com.congge.framework.Invocation;
import com.congge.framework.ProxyFactory;
import com.congge.framework.http.HttpClient;
import com.congge.service.HelloService;public class Consumer {public static void main(String[] args) {HttpClient httpClient = new HttpClient();Invocation invocation = new Invocation(HelloService.class.getName(),"sayHello",new Class[]{String.class},new Object[]{"jerry"});String result = httpClient.send("localhost", 8081, invocation);System.out.println(result);}}

启动消费端main程序后，执行服务调用，成功获取到服务提供方接口的响应结果

四、优化改进点

通过上面的案例代码的演示，简单实现了一个dubbo的调用过程，但这这是一个非常简单的实现，从生产者和消费端来看，均存在一些不太合理的地方，下面做几处简单的优化改进；

1、消费方改进

从上面的这一段调用来说，显得比较麻烦，很明显，这一整段代码可以通过一个类似代理工厂的方式进行封装，然后消费端只需要注入服务接口，调用服务接口的方法即可，这才是我们希望看到的；

自定义一个请求代理工厂

我们知道，在使用dubbo的时候，只需要注入服务接口即可，然后消费方就可以直接调用接口中的方法名了，但接口是不能执行的，最终需要一个代理类去执行接口的方法调用，容易想到的就是使用JDK的动态代理的方式来完成这件事，该类要做的事情就是这些；

这里我们先考虑用最简单的方式实现（先不考虑使用分布式注册中心）

import com.congge.framework.http.HttpClient;
import com.congge.framework.resgister.FileMapRegister;
import com.congge.service.HelloService;import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;public class ProxyFactory {@SuppressWarnings("unchecked")public static <T> T getProxy(Class interfaceClass) {return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {Invocation invocation = new Invocation(interfaceClass.getName(),method.getName(),method.getParameterTypes(),args);HttpClient httpClient = new HttpClient();//端口等信息的获取String result = httpClient.send("localhost", 8081, invocation);//聪注册中心获取//URL url = RemoteMapRegistry.get(interfaceClass.getName()).get(0);//方式一：Consumer从本地文件获取Provider地址//URL url = FileMapRegister.getURL(interfaceClass.getName());//方式二：Consumer从Zookeeper获取Provider地址//URL url = ZookeeperRegister.getURL(interfaceClass.getName());//String result = httpClient.send(url.getHostname(), url.getPort(), invocation);return result;}});}}

这样改进之后，那么在启动类中就可以像下面这样调用

HelloService helloService = ProxyFactory.getProxy(HelloService.class);String result = helloService.sayHello("zhangfei");System.out.println(result);

2、服务提供方改进

从上面的服务提供方一侧来看，存在一个比较明显的问题是，服务注册的时候，服务接口的注册并不是注册到分布式注册中心上面，由于 provider和consumer是两个JVM进程，consumer想要从pr注册中心拿到服务接口信息，这样是行不通的；

基于这个问题，从provider一侧来说，需要将服务注册到分布式注册中心上面去；

这里为了演示方便，我们提供两种改进方式，provider在启动的时候，将服务信息注册到本地的文件中，和注册到zk上；

提供一个URL类，记录provider的host和port信息

import java.io.Serializable;public class URL implements Serializable {private String hostname;private Integer port;public URL(String hostname, Integer port) {this.hostname = hostname;this.port = port;}public String getHostname() {return hostname;}public void setHostname(String hostname) {this.hostname = hostname;}public Integer getPort() {return port;}public void setPort(Integer port) {this.port = port;}@Overridepublic String toString() {return "URL{" +"hostname='" + hostname + '\'' +", port=" + port +'}';}}

提供一个FileMapRegister类，模拟将服务接口写入到本地文件

import java.io.*;
import java.util.HashMap;
import java.util.Map;public class FileMapRegister {private static Map<String, String> REGISTER = new HashMap<>();public static void regist(String interfaceName, String implClass, String url) {REGISTER.put(interfaceName, implClass + "::" + url);saveFile();}public static URL getURL(String interfaceName) {REGISTER = getFile();String[] s = REGISTER.get(interfaceName).split("::")[1].split(":");URL url = new URL(s[0],Integer.parseInt(s[1]));return url;}public static Class getImplClass(String interfaceName) throws ClassNotFoundException {REGISTER = getFile();return Class.forName(REGISTER.get(interfaceName).split("::")[0]);}public static void saveFile() {try {FileOutputStream fileOutputStream = new FileOutputStream("D:\\temp.txt");ObjectOutputStream objectOutputStream = new ObjectOutputStream(fileOutputStream);objectOutputStream.writeObject(REGISTER);} catch (IOException e) {e.printStackTrace();}}public static Map<String, String> getFile() {try {FileInputStream fileInputStream = new FileInputStream("D:\\temp.txt");ObjectInputStream objectInputStream = new ObjectInputStream(fileInputStream);return (Map<String, String>) objectInputStream.readObject();} catch (FileNotFoundException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();}return null;}}

提供一个ZookeeperRegister 类，可以将服务信息注册到zk

import com.congge.framework.URL;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;import java.util.HashMap;
import java.util.Map;
public class ZookeeperRegister {static CuratorFramework client;static Map<String, String> UrlCache = new HashMap<>();static {client = CuratorFrameworkFactory.newClient("localhost:2181", new RetryNTimes(3, 1000));client.start();}private static Map<String, String> REGISTER = new HashMap<>();//Provider注册服务public static void regist(String interfaceName, String implClass, String url) {try {Stat stat = client.checkExists().forPath(String.format("/dubbo/service/%s", interfaceName));if(stat != null){client.delete().forPath(String.format("/dubbo/service/%s", interfaceName));}String result = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(String.format("/dubbo/service/%s", interfaceName),(implClass + "::" + url).getBytes());System.out.println("Provier服务注册: " + result);} catch (Exception e) {e.printStackTrace();}}//获取Provider URLpublic static URL getURL(String interfaceName) {URL url = null;String urlString = null;//先查询缓存if (UrlCache.containsKey(interfaceName)) {urlString = UrlCache.get(interfaceName);} else {try {byte[] bytes = client.getData().forPath(String.format("/dubbo/service/%s", interfaceName));urlString = new String(bytes);} catch (Exception e) {e.printStackTrace();}}String host = urlString.split("::")[1].split(":")[0];String port = urlString.split("::")[1].split(":")[1];return new URL(host,Integer.parseInt(port));}//获取Provider实现类public static Class getImplClass(String interfaceName) throws Exception {byte[] bytes = client.getData().forPath(String.format("/dubbo/service/%s", interfaceName));String urlString = new String(bytes);return Class.forName(urlString.split("::")[0]);}
}

改造provider的启动类

服务启动时，将服务相关信息注册到本地文件

public class Provider {public static void main(String[] args) {//注册接口LocalRegister.register(HelloService.class.getName(),HelloServiceImpl.class);//注册服务的端口等信息//URL url = new URL("localhost",8081);//RemoteMapRegistry.register(HelloService.class.getName(), url);FileMapRegister.regist(HelloService.class.getName(),HelloServiceImpl.class.getName(),"localhost:8081");HttpServer httpServer = new HttpServer();httpServer.start("localhost",8081);}}

改造consumer端的ProxyFactory类

将从本地的文件中获取服务接口相关信息

public class ProxyFactory {@SuppressWarnings("unchecked")public static <T> T getProxy(Class interfaceClass) {return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {Invocation invocation = new Invocation(interfaceClass.getName(),method.getName(),method.getParameterTypes(),args);HttpClient httpClient = new HttpClient();//方式一：Consumer从本地文件获取Provider地址//URL url = FileMapRegister.getURL(interfaceClass.getName());//方式二：Consumer从Zookeeper获取Provider地址//URL url = ZookeeperRegister.getURL(interfaceClass.getName());String result = httpClient.send(url.getHostname(), url.getPort(), invocation);return result;}});}}

以上改造完毕后，再次启动provider和consumer，观察运行效果，consumer仍然可以正确拿到结果；

关于改造点，可以延续着这个思路继续进行下去，比如注册中心使用zk或redis，下面提几点以供参考：

消费端的服务调用容错处理，假如调用接口超时？或者异常？
消费端的服务重试；
消费端的负载均衡策略如何指定？
将嵌入式tomcat容器改为netty；