devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

阿里巴巴中间件性能挑战赛（RPC篇 同步阻塞模型）_com.alibaba.middleware

作者：我家小花儿 | 2024-04-14 23:13:00

赞

踩

com.alibaba.middleware

赛题要求：

一个简单的RPC框架

RPC（Remote Procedure Call ）——远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

框架——让编程人员便捷地使用框架所提供的功能，由于RPC的特性，聚焦于应用的分布式服务化开发，所以成为一个对开发人员无感知的接口代理，显然是RPC框架优秀的设计。

题目要求

1.要成为框架：对于框架的使用者，隐藏RPC实现。

2.网络模块可以自己编写，如果要使用IO框架，要求使用netty-4.0.23.Final。

3.支持异步调用，提供future、callback的能力。

4.能够传输基本类型、自定义业务类型、异常类型（要在客户端抛出）。

5.要处理超时场景，服务端处理时间较长时，客户端在指定时间内跳出本次调用。

6.提供RPC上下文，客户端可以透传数据给服务端。

7.提供Hook，让开发人员进行RPC层面的AOP。

注：为了降低第一题的难度，RPC框架不需要注册中心，客户端识别-DSIP的JVM参数来获取服务端IP。

衡量标准

满足所有要求。性能测试。



这个赛题现在看来其实并不难，不过当时刚参赛的时候我还是理解了好久。首先刚开始的时候其实根本不知道到底什么是RPC，感觉应该是一种无比高大上无比难实现的东东。直到我看到了梁飞的博客http://javatar.iteye.com/blog/1123915，才知道原来RPC其实并不难。通俗来说就是客户端通过一定的协议把方法名称、参数类型还有参数传给服务器，然后服务器调用对应方法，完成以后再把结果传回来给客户端，仅此而已。



首先是Consumer部分的接口定义：

  1. package com.alibaba.middleware.race.rpc.api;
  2.  
  3. import com.alibaba.middleware.race.rpc.aop.ConsumerHook;
  4. import com.alibaba.middleware.race.rpc.async.ResponseCallbackListener;
  5.  
  6. import java.lang.reflect.InvocationHandler;
  7. import java.lang.reflect.Method;
  8. import java.lang.reflect.Proxy;
  9.  
  10. /**
  11.  * Created by huangsheng.hs on 2015/3/26.
  12.  */
  13. public class RpcConsumer implements InvocationHandler {
  14.     private Class<?> interfaceClazz;
  15.     public RpcConsumer() {
  16.     }
  17.  
  18.     private void init() {
  19.         //TODO
  20.     }
  21.  
  22.  
  23.     public RpcConsumer interfaceClass(Class<?> interfaceClass) {
  24.         this.interfaceClazz = interfaceClass;
  25.         return this;
  26.     }
  27.  
  28.     public RpcConsumer version(String version) {
  29.         //TODO
  30.         return this;
  31.     }
  32.  
  33.  
  34.     public RpcConsumer clientTimeout(int clientTimeout) {
  35.         //TODO
  36.         return this;
  37.     }
  38.  
  39.  
  40.     public RpcConsumer hook(ConsumerHook hook) {
  41.         return this;
  42.     }
  43.  
  44.     public Object instance() {
  45.         //TODO return an Proxy
  46.         return Proxy.newProxyInstance(this.getClass().getClassLoader(),new Class[]{this.interfaceClazz},this);
  47.     }
  48.  
  49.     public void asynCall(String methodName) {
  50.         asynCall(methodName, null);
  51.     }
  52.  
  53.     public <T extends ResponseCallbackListener> void asynCall(String methodName, T callbackListener) {
  54.         //TODO
  55.     }
  56.  
  57.     public void cancelAsyn(String methodName) {
  58.         //TODO
  59.     }
  60.  
  61.  
  62.     @Override
  63.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  64.         return null;
  65.     }
  66. }
  67.
主要接口分析如下

public RpcConsumer interfaceClass(Class<?> interfaceClass)：定义了RPC的服务接口（也就是客户端可以调用服务器提供的哪些服务）

public Object instance():生成一个客户端代理的实例，其中用到了Proxy类的newProxyInstance方法

public asynCall(String methodName,T callbackListener):异步调用的接口，先在注册表中注册异步回调函数，然后进行异步调用

public Object invoke(Object proxy, Method method, Object[] args):RpcConsumer实现了java中的InvocationHandler接口，因此需要重写invoke方法。invoke方法是RpcConsumer最核心的方法。调用被代理的任何服务都会被转为调用invoke方法，在invoke方法中会向服务器发送Rpc请求。如果是同步调用则会阻塞等待结果，如果是异步调用则会马上返回。

（invoke方法是程序最核心的地方，也是比较难理解的地方。如果对于invoke方法不理解，可以先参考java的代理类的原理，了解Proxy类以及InvocationHandler接口的原理再往下看）


接下来是RpcConsumer的具体实现（只贴最核心的invoke方法，其余代码请看我的github）

  1. //调用方法的协议：方法名  参数类型  参数值  上下文
  2.     @Override
  3.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable 
  4.     {
  5.     	if(isAsyn(method.getName())) return null; //本方法是同步方法，被异步调用则返回空
  6.     	Socket s = new Socket(host,PORT);
  7.     	try
  8.     	{
  9.     		ObjectOutputStream output = new ObjectOutputStream(s.getOutputStream());
  10.     		try
  11.     		{
  12.     			RpcRequest request = new RpcRequest(method.getName(), method.getParameterTypes(), 
  13.     					args, RpcContext.localMap.get());
  14.     			consumerHook.before(request);
  15.     			output.writeObject(request);
  16.     			ObjectInputStream input = new ObjectInputStream(s.getInputStream());
  17.     			try
  18.     			{	
  19.     				//使用Callable接口和FutureTask来实现计时
  20.     				FutureTask<RpcResponse> task = new FutureTask<RpcResponse>(new ResultGetter(input));
  21.     				new Thread(task).start();
  22.     				RpcResponse response= task.get(clientTimeout, TimeUnit.MILLISECONDS);
  23.     				if(response.isError())
  24.     				{
  25.     					InvocationTargetException targetEx =  (InvocationTargetException)(response.getThrowable());
  26.     					Throwable t = targetEx .getTargetException();
  27.     					throw t;
  28.     				}
  29.     				else
  30.     				{
  31.     					Object result = response.getAppResponse();
  32.     					consumerHook.after(request);
  33.         				return result;
  34.     				}	
  35.     				
  36.     			}
  37.     			finally
  38.     			{
  39.     				input.close();
  40.     			}
  41.     		}
  42.     		finally
  43.     		{
  44.     			output.close();
  45.     		}
  46.     	}
  47.     	finally
  48.     	{
  49.     		s.close();
  50.     	}
  51.     }
思路其实很简单，就是把方法名、参数类型、参数值以及其他东东包装成RpcRequest，然后用java自带的序列化方法通过socket发送到服务器端，然后再同步阻塞等待结果返回回来。当然，这个只是最简单的版本，后面我会将一下一些更加复杂的细节。不过在此之前先贴一下服务器端的代码：


接口定义：

  1. package com.alibaba.middleware.race.rpc.api;
  2.  
  3. /**
  4.  * Created by huangsheng.hs on 2015/3/26.
  5.  */
  6. public class RpcProvider {
  7.  
  8.     public RpcProvider() {}
  9.  
  10.     /**
  11.      * init Provider
  12.      */
  13.     private void init(){
  14.         //TODO
  15.     }
  16.  
  17.     /**
  18.      * set the interface which this provider want to expose as a service
  19.      * @param serviceInterface
  20.      */
  21.     public RpcProvider serviceInterface(Class<?> serviceInterface){
  22.         //TODO
  23.         return this;
  24.     }
  25.  
  26.     /**
  27.      * set the version of the service
  28.      * @param version
  29.      */
  30.     public RpcProvider version(String version){
  31.         //TODO
  32.         return this;
  33.     }
  34.  
  35.     /**
  36.      * set the instance which implements the service's interface
  37.      * @param serviceInstance
  38.      */
  39.     public RpcProvider impl(Object serviceInstance){
  40.         //TODO
  41.         return this;
  42.     }
  43.  
  44.     /**
  45.      * set the timeout of the service
  46.      * @param timeout
  47.      */
  48.     public RpcProvider timeout(int timeout){
  49.         //TODO
  50.         return this;
  51.     }
  52.  
  53.     /**
  54.      * set serialize type of this service
  55.      * @param serializeType
  56.      */
  57.     public RpcProvider serializeType(String serializeType){
  58.         //TODO
  59.         return this;
  60.     }
  61.  
  62.     /**
  63.      * publish this service
  64.      * if you want to publish your service , you need a registry server.
  65.      * after all , you cannot write servers' ips in config file when you have 1 million server.
  66.      * you can use ZooKeeper as your registry server to make your services found by your consumers.
  67.      */
  68.     public void publish() {
  69.         //TODO
  70.     }
  71. }
具体实现（其他方法都不是很重要，只贴publish方法）： 

  1. public void publish(Object service,int port) 
  2.     {
  3.     	System.out.println("service:"+service.getClass().getName()+"  port:"+port);
  4.     	
  5.     	ServerSocket ss = null;
  6. 		try {
  7. 			ss = new ServerSocket(port);
  8. 			while(true)
  9. 	    	{
  10. 				
  11. 	    		final Socket socket = ss.accept();
  12. 	    		Thread thread = new Thread(new ServiceThread(socket,serviceInstance)); 	
  13. 	    		thread.start();
  14. 	    		try{
  15. 	    			thread.join(TIME_OUT);
  16. 	    		}catch(InterruptedException e){
  17. 	    			e.printStackTrace();
  18. 	    			System.out.println("thread被打断");
  19. 	    		}
  20. 	    		if(thread.isAlive())
  21. 	    		{
  22. 	    			thread.interrupt();
  23. 	    		}
  24. 	    	}
  25. 		} catch (Exception e) {
  26. 			System.out.println("thread被打断");
  27. 			e.printStackTrace();
  28. 		}
  29.     	
  30.     }
服务端的思路其实也很简单，就是起一个简单的socket,不断地监听在某一个端口上，如果有新的请求到来就新起一个线程（ServiceThread）来处理请求


ServiceThread代码如下：

  1. package com.alibaba.middleware.race.rpc.api.impl;
  2.  
  3. import java.io.*;
  4. import java.lang.reflect.Method;
  5. import java.net.Socket;
  6. import java.util.Map;
  7. import java.util.Set;
  8. import java.util.concurrent.ExecutorService;
  9. import java.util.concurrent.Executors;
  10. import java.util.concurrent.Future;
  11. import java.util.concurrent.TimeoutException;
  12.  
  13. import com.alibaba.middleware.race.rpc.context.RpcContext;
  14. import com.alibaba.middleware.race.rpc.model.RpcRequest;
  15. import com.alibaba.middleware.race.rpc.model.RpcResponse;
  16.  
  17.  
  18. //RpcContext需要对props实现ThreadLocal以保证线程的同步访问
  19. public class ServiceThread implements Runnable{
  20. 	
  21. 	public ServiceThread(final Socket s,final Object service)
  22. 	{
  23. 		this.s = s;
  24. 		this.service = service;
  25. 	}
  26. 	
  27. 	@Override
  28. 	public void run() 
  29. 	{
  30. 		try
  31. 		{
  32. 			try
  33. 			{
  34. 				ObjectInputStream input = new ObjectInputStream(s.getInputStream()); 
  35. 				
  36. 				try
  37. 				{
  38. 					RpcRequest request = (RpcRequest)input.readObject();
  39. 					RpcContext.localMap.set(request.getContextMap());
  40. 					ObjectOutputStream output = new ObjectOutputStream(s.getOutputStream());
  41. 					Method method = service.getClass().getMethod(request.getMethodName(), 
  42. 											request.getParameterTypes());
  43. 					RpcResponse response = new RpcResponse();
  44. 					try
  45. 					{
  46. 						Object result = method.invoke(service, request.getArgs());
  47. 						response.setAppResponse(result);
  48. 						output.writeObject(response);
  49. 					}catch(Throwable t)
  50. 					{	
  51. 						response.setThrowable(t);
  52. 						output.writeObject(response);
  53. 					}finally
  54. 					{
  55. 						output.close();
  56. 					}
  57. 				}
  58. 				finally
  59. 				{
  60. 					input.close();
  61. 				}
  62. 				
  63. 			}
  64. 			finally
  65. 			{
  66. 				s.close();
  67. 			}
  68. 		}catch(Exception e)
  69. 		{
  70. 			e.printStackTrace();
  71. 		}
  72. 		
  73. 	}
  74. 	final Socket s;
  75. 	final Object service;
  76. }

基本的原理大概就是这样，最后再讲一下一些实现的细节以及优化：

1、如何实现异步调用：由于java没有函数指针，因此可以通过在注册表中注册回调接口的方式来实现。

2、对于更加复杂的RPC框架，会使用Zookeeper来实现注册中心。

3、在程序中使用的网络io模型是同步阻塞的io模型，在比赛中，我还基于netty框架尝试了同步非阻塞的网络io模型，这个会在接下来再细讲。

4、序列化的方法使用的是java自带的序列化，这其实是十分低效的一种方式。实际上可以使用kryo、protobuf、fst等更加高效的序列库。



声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/424594
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号