轻量级分布式RPC框架实现(续)

时间:2023-02-06 20:36:39
1、背景

最近在搜索Netty和Zookeeper方面的文章时,看到了这篇文章《轻量级分布式 RPC 框架》,作者用Zookeeper、Netty和Spring写了一个轻量级的分布式RPC框架。花了一些时间看了下他的代码,写的干净简单,写的RPC框架可以算是一个简易版的dubbo。这个RPC框架虽小,但是麻雀虽小,五脏俱全,有兴趣的可以学习一下。

本人在这个简易版的RPC上添加了如下特性:

* 服务异步调用的支持,回调函数callback的支持

* 客户端使用长连接(在多次调用共享连接)

* 服务端异步多线程处理RPC请求

项目地址:https://github.com/luxiaoxun/NettyRpc

2、简介

RPC,即 Remote Procedure Call(远程过程调用),调用远程计算机上的服务,就像调用本地服务一样。RPC可以很好的解耦系统,如WebService就是一种基于Http协议的RPC。

这个RPC整体框架如下:

轻量级分布式RPC框架实现(续)

这个RPC框架使用的一些技术所解决的问题:

服务发布与订阅:服务端使用Zookeeper注册服务地址,客户端从Zookeeper获取可用的服务地址。

通信:使用Netty作为通信框架。

Spring:使用Spring配置服务,加载Bean,扫描注解。

动态代理:客户端使用代理模式透明化服务调用。

消息编解码:使用Protostuff序列化和反序列化消息。

3、服务端发布服务

使用注解标注要发布的服务

服务注解

  1. @Target({ElementType.TYPE})  
  2. @Retention(RetentionPolicy.RUNTIME)  
  3. @Component  
  4. public @interface RpcService {  
  5.     Class<?> value();  
  6. }  
轻量级分布式RPC框架实现(续)

一个服务接口:

  1. public interface HelloService {  
  2.   
  3.     String hello(String name);  
  4.   
  5.     String hello(Person person);  
  6. }  
轻量级分布式RPC框架实现(续)
一个服务实现:使用注解标注

  1. @RpcService(HelloService.class)  
  2. public class HelloServiceImpl implements HelloService {  
  3.   
  4.     @Override  
  5.     public String hello(String name) {  
  6.         return "Hello! " + name;  
  7.     }  
  8.   
  9.     @Override  
  10.     public String hello(Person person) {  
  11.         return "Hello! " + person.getFirstName() + " " + person.getLastName();  
  12.     }  
  13. }  
轻量级分布式RPC框架实现(续)

服务在启动的时候扫描得到所有的服务接口及其实现:

  1. @Override  
  2.     public void setApplicationContext(ApplicationContext ctx) throws BeansException {  
  3.         Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class);  
  4.         if (MapUtils.isNotEmpty(serviceBeanMap)) {  
  5.             for (Object serviceBean : serviceBeanMap.values()) {  
  6.                 String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName();  
  7.                 handlerMap.put(interfaceName, serviceBean);  
  8.             }  
  9.         }  
  10.     }  
轻量级分布式RPC框架实现(续)

在Zookeeper集群上注册服务地址:

  1. public class ServiceRegistry {  
  2.   
  3.     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceRegistry.class);  
  4.   
  5.     private CountDownLatch latch = new CountDownLatch(1);  
  6.   
  7.     private String registryAddress;  
  8.   
  9.     public ServiceRegistry(String registryAddress) {  
  10.         this.registryAddress = registryAddress;  
  11.     }  
  12.   
  13.     public void register(String data) {  
  14.         if (data != null) {  
  15.             ZooKeeper zk = connectServer();  
  16.             if (zk != null) {  
  17.                 AddRootNode(zk); // Add root node if not exist  
  18.                 createNode(zk, data);  
  19.             }  
  20.         }  
  21.     }  
  22.   
  23.     private ZooKeeper connectServer() {  
  24.         ZooKeeper zk = null;  
  25.         try {  
  26.             zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {  
  27.                 @Override  
  28.                 public void process(WatchedEvent event) {  
  29.                     if (event.getState() == Event.KeeperState.SyncConnected) {  
  30.                         latch.countDown();  
  31.                     }  
  32.                 }  
  33.             });  
  34.             latch.await();  
  35.         } catch (IOException e) {  
  36.             LOGGER.error("", e);  
  37.         }  
  38.         catch (InterruptedException ex){  
  39.             LOGGER.error("", ex);  
  40.         }  
  41.         return zk;  
  42.     }  
  43.   
  44.     private void AddRootNode(ZooKeeper zk){  
  45.         try {  
  46.             Stat s = zk.exists(Constant.ZK_REGISTRY_PATH, false);  
  47.             if (s == null) {  
  48.                 zk.create(Constant.ZK_REGISTRY_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
  49.             }  
  50.         } catch (KeeperException e) {  
  51.             LOGGER.error(e.toString());  
  52.         } catch (InterruptedException e) {  
  53.             LOGGER.error(e.toString());  
  54.         }  
  55.     }  
  56.   
  57.     private void createNode(ZooKeeper zk, String data) {  
  58.         try {  
  59.             byte[] bytes = data.getBytes();  
  60.             String path = zk.create(Constant.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);  
  61.             LOGGER.debug("create zookeeper node ({} => {})", path, data);  
  62.         } catch (KeeperException e) {  
  63.             LOGGER.error("", e);  
  64.         }  
  65.         catch (InterruptedException ex){  
  66.             LOGGER.error("", ex);  
  67.         }  
  68.     }  
  69. }  
轻量级分布式RPC框架实现(续)

这里在原文的基础上加了AddRootNode()判断服务父节点是否存在,如果不存在则添加一个PERSISTENT的服务父节点,这样虽然启动服务时多了点判断,但是不需要手动命令添加服务父节点了。

关于Zookeeper的使用原理,可以看这里《ZooKeeper基本原理》。

4、客户端调用服务

使用代理模式调用服务:

  1. public class RpcProxy {  
  2.   
  3.     private String serverAddress;  
  4.     private ServiceDiscovery serviceDiscovery;  
  5.   
  6.     public RpcProxy(String serverAddress) {  
  7.         this.serverAddress = serverAddress;  
  8.     }  
  9.   
  10.     public RpcProxy(ServiceDiscovery serviceDiscovery) {  
  11.         this.serviceDiscovery = serviceDiscovery;  
  12.     }  
  13.   
  14.     @SuppressWarnings("unchecked")  
  15.     public <T> T create(Class<?> interfaceClass) {  
  16.         return (T) Proxy.newProxyInstance(  
  17.                 interfaceClass.getClassLoader(),  
  18.                 new Class<?>[]{interfaceClass},  
  19.                 new InvocationHandler() {  
  20.                     @Override  
  21.                     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {  
  22.                         RpcRequest request = new RpcRequest();  
  23.                         request.setRequestId(UUID.randomUUID().toString());  
  24.                         request.setClassName(method.getDeclaringClass().getName());  
  25.                         request.setMethodName(method.getName());  
  26.                         request.setParameterTypes(method.getParameterTypes());  
  27.                         request.setParameters(args);  
  28.   
  29.                         if (serviceDiscovery != null) {  
  30.                             serverAddress = serviceDiscovery.discover();  
  31.                         }  
  32.                         if(serverAddress != null){  
  33.                             String[] array = serverAddress.split(":");  
  34.                             String host = array[0];  
  35.                             int port = Integer.parseInt(array[1]);  
  36.   
  37.                             RpcClient client = new RpcClient(host, port);  
  38.                             RpcResponse response = client.send(request);  
  39.   
  40.                             if (response.isError()) {  
  41.                                 throw new RuntimeException("Response error.",new Throwable(response.getError()));  
  42.                             } else {  
  43.                                 return response.getResult();  
  44.                             }  
  45.                         }  
  46.                         else{  
  47.                             throw new RuntimeException("No server address found!");  
  48.                         }  
  49.                     }  
  50.                 }  
  51.         );  
  52.     }  
  53. }  
轻量级分布式RPC框架实现(续)

这里每次使用代理远程调用服务,从Zookeeper上获取可用的服务地址,通过RpcClient send一个Request,等待该Request的Response返回。这里原文有个比较严重的bug,在原文给出的简单的Test中是很难测出来的,原文使用了obj的wait和notifyAll来等待Response返回,会出现“假死等待”的情况:一个Request发送出去后,在obj.wait()调用之前可能Response就返回了,这时候在channelRead0里已经拿到了Response并且obj.notifyAll()已经在obj.wait()之前调用了,这时候send后再obj.wait()就出现了假死等待,客户端就一直等待在这里。使用CountDownLatch可以解决这个问题。

注意:这里每次调用的send时候才去和服务端建立连接,使用的是短连接,这种短连接在高并发时会有连接数问题,也会影响性能。

从Zookeeper上获取服务地址:


  1. public class ServiceDiscovery {  
  2.   
  3.     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceDiscovery.class);  
  4.   
  5.     private CountDownLatch latch = new CountDownLatch(1);  
  6.   
  7.     private volatile List<String> dataList = new ArrayList<>();  
  8.   
  9.     private String registryAddress;  
  10.   
  11.     public ServiceDiscovery(String registryAddress) {  
  12.         this.registryAddress = registryAddress;  
  13.         ZooKeeper zk = connectServer();  
  14.         if (zk != null) {  
  15.             watchNode(zk);  
  16.         }  
  17.     }  
  18.   
  19.     public String discover() {  
  20.         String data = null;  
  21.         int size = dataList.size();  
  22.         if (size > 0) {  
  23.             if (size == 1) {  
  24.                 data = dataList.get(0);  
  25.                 LOGGER.debug("using only data: {}", data);  
  26.             } else {  
  27.                 data = dataList.get(ThreadLocalRandom.current().nextInt(size));  
  28.                 LOGGER.debug("using random data: {}", data);  
  29.             }  
  30.         }  
  31.         return data;  
  32.     }  
  33.   
  34.     private ZooKeeper connectServer() {  
  35.         ZooKeeper zk = null;  
  36.         try {  
  37.             zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {  
  38.                 @Override  
  39.                 public void process(WatchedEvent event) {  
  40.                     if (event.getState() == Event.KeeperState.SyncConnected) {  
  41.                         latch.countDown();  
  42.                     }  
  43.                 }  
  44.             });  
  45.             latch.await();  
  46.         } catch (IOException | InterruptedException e) {  
  47.             LOGGER.error("", e);  
  48.         }  
  49.         return zk;  
  50.     }  
  51.   
  52.     private void watchNode(final ZooKeeper zk) {  
  53.         try {  
  54.             List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, new Watcher() {  
  55.                 @Override  
  56.                 public void process(WatchedEvent event) {  
  57.                     if (event.getType() == Event.EventType.NodeChildrenChanged) {  
  58.                         watchNode(zk);  
  59.                     }  
  60.                 }  
  61.             });  
  62.             List<String> dataList = new ArrayList<>();  
  63.             for (String node : nodeList) {  
  64.                 byte[] bytes = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, falsenull);  
  65.                 dataList.add(new String(bytes));  
  66.             }  
  67.             LOGGER.debug("node data: {}", dataList);  
  68.             this.dataList = dataList;  
  69.         } catch (KeeperException | InterruptedException e) {  
  70.             LOGGER.error("", e);  
  71.         }  
  72.     }  
  73. }  
轻量级分布式RPC框架实现(续)

每次服务地址节点发生变化,都需要再次watchNode,获取新的服务地址列表。

5、消息编码

请求消息:

  1. public class RpcRequest {  
  2.   
  3.     private String requestId;  
  4.     private String className;  
  5.     private String methodName;  
  6.     private Class<?>[] parameterTypes;  
  7.     private Object[] parameters;  
  8.   
  9.     public String getRequestId() {  
  10.         return requestId;  
  11.     }  
  12.   
  13.     public void setRequestId(String requestId) {  
  14.         this.requestId = requestId;  
  15.     }  
  16.   
  17.     public String getClassName() {  
  18.         return className;  
  19.     }  
  20.   
  21.     public void setClassName(String className) {  
  22.         this.className = className;  
  23.     }  
  24.   
  25.     public String getMethodName() {  
  26.         return methodName;  
  27.     }  
  28.   
  29.     public void setMethodName(String methodName) {  
  30.         this.methodName = methodName;  
  31.     }  
  32.   
  33.     public Class<?>[] getParameterTypes() {  
  34.         return parameterTypes;  
  35.     }  
  36.   
  37.     public void setParameterTypes(Class<?>[] parameterTypes) {  
  38.         this.parameterTypes = parameterTypes;  
  39.     }  
  40.   
  41.     public Object[] getParameters() {  
  42.         return parameters;  
  43.     }  
  44.   
  45.     public void setParameters(Object[] parameters) {  
  46.         this.parameters = parameters;  
  47.     }  
  48. }  
轻量级分布式RPC框架实现(续)
响应消息:

  1. public class RpcResponse {  
  2.   
  3.     private String requestId;  
  4.     private String error;  
  5.     private Object result;  
  6.   
  7.     public boolean isError() {  
  8.         return error != null;  
  9.     }  
  10.   
  11.     public String getRequestId() {  
  12.         return requestId;  
  13.     }  
  14.   
  15.     public void setRequestId(String requestId) {  
  16.         this.requestId = requestId;  
  17.     }  
  18.   
  19.     public String getError() {  
  20.         return error;  
  21.     }  
  22.   
  23.     public void setError(String error) {  
  24.         this.error = error;  
  25.     }  
  26.   
  27.     public Object getResult() {  
  28.         return result;  
  29.     }  
  30.   
  31.     public void setResult(Object result) {  
  32.         this.result = result;  
  33.     }  
  34. }  
轻量级分布式RPC框架实现(续)

消息序列化和反序列化工具:(基于 Protostuff 实现)

  1. public class SerializationUtil {  
  2.   
  3.     private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<>();  
  4.   
  5.     private static Objenesis objenesis = new ObjenesisStd(true);  
  6.   
  7.     private SerializationUtil() {  
  8.     }  
  9.   
  10.     @SuppressWarnings("unchecked")  
  11.     private static <T> Schema<T> getSchema(Class<T> cls) {  
  12.         Schema<T> schema = (Schema<T>) cachedSchema.get(cls);  
  13.         if (schema == null) {  
  14.             schema = RuntimeSchema.createFrom(cls);  
  15.             if (schema != null) {  
  16.                 cachedSchema.put(cls, schema);  
  17.             }  
  18.         }  
  19.         return schema;  
  20.     }  
  21.   
  22.     /** 
  23.      * 序列化(对象 -> 字节数组) 
  24.      */  
  25.     @SuppressWarnings("unchecked")  
  26.     public static <T> byte[] serialize(T obj) {  
  27.         Class<T> cls = (Class<T>) obj.getClass();  
  28.         LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);  
  29.         try {  
  30.             Schema<T> schema = getSchema(cls);  
  31.             return ProtostuffIOUtil.toByteArray(obj, schema, buffer);  
  32.         } catch (Exception e) {  
  33.             throw new IllegalStateException(e.getMessage(), e);  
  34.         } finally {  
  35.             buffer.clear();  
  36.         }  
  37.     }  
  38.   
  39.     /** 
  40.      * 反序列化(字节数组 -> 对象) 
  41.      */  
  42.     public static <T> T deserialize(byte[] data, Class<T> cls) {  
  43.         try {  
  44.             T message = (T) objenesis.newInstance(cls);  
  45.             Schema<T> schema = getSchema(cls);  
  46.             ProtostuffIOUtil.mergeFrom(data, message, schema);  
  47.             return message;  
  48.         } catch (Exception e) {  
  49.             throw new IllegalStateException(e.getMessage(), e);  
  50.         }  
  51.     }  
  52. }  
轻量级分布式RPC框架实现(续)

由于处理的是TCP消息,本人加了TCP的粘包处理Handler

  1. channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(65536,0,4,0,0))  
轻量级分布式RPC框架实现(续)

消息编解码时开始4个字节表示消息的长度,也就是消息编码的时候,先写消息的长度,再写消息。

6、性能改进

1)服务端请求异步处理

Netty本身就是一个高性能的网络框架,从网络IO方面来说并没有太大的问题。

从这个RPC框架本身来说,在原文的基础上把Server端处理请求的过程改成了多线程异步:

  1. public void channelRead0(final ChannelHandlerContext ctx,final RpcRequest request) throws Exception {  
  2.        RpcServer.submit(new Runnable() {  
  3.            @Override  
  4.            public void run() {  
  5.                LOGGER.debug("Receive request " + request.getRequestId());  
  6.                RpcResponse response = new RpcResponse();  
  7.                response.setRequestId(request.getRequestId());  
  8.                try {  
  9.                    Object result = handle(request);  
  10.                    response.setResult(result);  
  11.                } catch (Throwable t) {  
  12.                    response.setError(t.toString());  
  13.                    LOGGER.error("RPC Server handle request error",t);  
  14.                }  
  15.                ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE).addListener(new ChannelFutureListener() {  
  16.                    @Override  
  17.                    public void operationComplete(ChannelFuture channelFuture) throws Exception {  
  18.                        LOGGER.debug("Send response for request " + request.getRequestId());  
  19.                    }  
  20.                });  
  21.            }  
  22.        });  
  23.    }  
轻量级分布式RPC框架实现(续)

Netty 4中的Handler处理在IO线程中,如果Handler处理中有耗时的操作(如数据库相关),会让IO线程等待,影响性能。

2)服务端长连接的管理

 客户端保持和服务进行长连接,不需要每次调用服务的时候进行连接,长连接的管理(通过Zookeeper获取有效的地址)。

通过监听Zookeeper服务节点值的变化,动态更新客户端和服务端保持的长连接。这个事情现在放在客户端在做,客户端保持了和所有可用服务的长连接,给客户端和服务端都造成了压力,需要解耦这个实现。

3)客户端请求异步处理

客户端请求异步处理的支持,不需要同步等待:发送一个异步请求,返回Feature,通过Feature的callback机制获取结果。


  1. IAsyncObjectProxy client = rpcClient.createAsync(HelloService.class);  
  2. RPCFuture helloFuture = client.call("hello", Integer.toString(i));  
  3. String result = (String) helloFuture.get(3000, TimeUnit.MILLISECONDS);  
  4.    
轻量级分布式RPC框架实现(续)

个人觉得该RPC的待改进项:

* 编码序列化的多协议支持。

 

项目持续更新中。

项目地址:https://github.com/luxiaoxun/NettyRpc

 

参考:

轻量级分布式 RPC 框架:http://my.oschina.net/huangyong/blog/361751

你应该知道的RPC原理:http://www.cnblogs.com/LBSer/p/4853234.html