手写RPC框架

 

源码地址：https://github.com/ItGeneral/code-framework

微信公众号原文：https://mp.weixin.qq.com/s/z1kIqvlX92oun1Cpm71Tsw

 

RPC(Remote Procedure Call Protocol)远程过程调用协议，采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息的到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。使用RPC框架，可以让远程接口调用像调用本地方法一样方便。 下面介绍一个简单的RPC框架，让读者领略RPC框架的原理

RPC服务端

（1）新增注解@RpcExporter，接口上添加该注解，即表示为RPC服务端接口

1. @Target(ElementType.METHOD)
2. @Retention(RetentionPolicy.RUNTIME)
3. @Documented
4. public @interface RpcExporter {
6. /**
7. * RPC服务端处理器名称（接口名称）
8. * @return
9. */
10. String value();
12. }

（2）新增接口RpcTestService及其实现RpcTestServiceImpl

1. public interface RpcTestService {
2. Boolean testRpcService(String params);
3. }
5. @Service
6. public class RpcTestServiceImpl implements RpcTestService{
7. private final static Logger LOGGER = getLogger(RpcTestServiceImpl.class);
9. @Override
10. @RpcExporter(value = "testRpcService")
11. public Boolean testRpcService(String params){
12. LOGGER.info("======testRpcService====success====" + params);
13. return true;
14. }
15. }

RPC客户端

（1）新增注解@RpcClient，接口方法上添加该注解，即表示为RPC客户端接口

1. @Target(ElementType.METHOD)
2. @Retention(RetentionPolicy.RUNTIME)
3. @Documented
4. public @interface RpcClient {
6. /**
7. * 对应的服务端处理器名称（bean名称+接口名称）
8. * @return
9. */
10. String value();
12. /**
13. * rpc 服务端地址
14. * @return
15. */
16. String rpcExportUrl();
17. }

（2）添加@RpcInterface注解，接口类上添加此注解，表明为RPC客户端接口类

1. @Target(ElementType.TYPE)
2. @Retention(RetentionPolicy.RUNTIME)
3. @Documented
4. public @interface RpcInterface {
6. }

（3）新增RPC客户端接口

1. @Service
2. @RpcInterface
3. public interface RpcTestClient {
5. @RpcClient(value = "RpcTestServiceImpl.testRpcService", rpcExportUrl = "http://localhost:8080/rpc")
6. Boolean testClient(String params);
8. }

初始化RPC框架

（1）添加@EnableRPC注解，开启RPC配置

1. @Documented
2. @Configuration
3. @Target(ElementType.TYPE)
4. @Retention(RetentionPolicy.RUNTIME)
5. @Import({RpcConfiguration.RpcPackageRegister.class, RpcConfiguration.class})
6. public @interface EnableRPC {
8. /**
9. * RPC 扫描包路径
10. * @return
11. */
12. String[] basePackages() default {};
14. }
16. public class RpcConfiguration {
18. private static Set<String> basePackages = new LinkedHashSet<>();
20. /**
21. * basePackages路径下的rpc客户端接口，设置为动态代理
22. * @return
23. */
24. @Bean
25. public RpcRegister rpcRegister(){
26. return new RpcRegister(basePackages);
27. }
29. public static class RpcPackageRegister implements ImportBeanDefinitionRegistrar{
30. /**
31. * 解析EnableRPC basePackages参数
32. * @param metadata
33. * @param registry
34. */
35. @Override
36. public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry registry) {
37. Map<String, Object> enableMap = metadata.getAnnotationAttributes(EnableRPC.class.getName());
38. AnnotationAttributes annotationAttributes = AnnotationAttributes.fromMap(enableMap);
39. String[] packages = annotationAttributes.getStringArray("basePackages");
40. for (String basePackage : packages){
41. String[] tokenize = StringUtils.tokenizeToStringArray(basePackage,
42. ConfigurableApplicationContext.CONFIG_LOCATION_DELIMITERS);
43. basePackages.addAll(Arrays.asList(tokenize));
44. }
45. if (basePackages.isEmpty()) {
46. basePackages.add(ClassUtils.getPackageName(metadata.getClassName()));
47. }
48. }
49. }
50. }

（2）spring bean 初始化前为所有RPC客户端接口设置动态代理，自定义invoke方法实现RPC客户端向RPC服务端请求数据的逻辑

1. public class RpcRegister implements BeanFactoryPostProcessor {
3. /**
4. * rpc扫描包
5. */
6. private Set<String> baseScanPackages;
8. public RpcRegister(Set<String> baseScanPackages) {
9. this.baseScanPackages = baseScanPackages;
10. }
12. /**
13. * spring容器在初始化bean之前，允许自定义某些bean
14. * RPC客户端所有bean设置动态代理，调用时走动态代理的invoke()
15. * @param beanFactory
16. * @throws BeansException
17. */
18. @Override
19. public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
20. DefaultListableBeanFactory factory = (DefaultListableBeanFactory) beanFactory;
21. for (String basePackage : baseScanPackages){
22. try {
23. Set<Class<?>> classSet = scanBackPackage(basePackage);
24. for (Class<?> clazz : classSet){
25. if (clazz.isInterface() && clazz.isAnnotationPresent(RpcInterface.class)){
26. //生成动态代理
27. Object proxy = RpcClientProxy.getProxy(clazz);
28. //将bean注入到spring容器中
29. factory.registerSingleton(clazz.getName(), proxy);
30. }
31. }
32. } catch (Exception e) {
33. e.printStackTrace();
34. }
35. }
36. }
39. /**
40. * 扫描包下面的所有类
41. * @param basePackage
42. * @return
43. */
44. private Set<Class<?>> scanBackPackage(String basePackage){
45. Set<Class<?>> classSet = new HashSet<>();
46. try{
47. String pattern = "classpath*:" + ClassUtils.convertClassNameToResourcePath(basePackage) + "/**/*.class";
48. ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
49. Resource[] resources = resourcePatternResolver.getResources(pattern);
50. MetadataReaderFactory readerFactory = new CachingMetadataReaderFactory(resourcePatternResolver);
51. for (Resource resource : resources){
52. if (resource.isReadable()){
53. MetadataReader reader = readerFactory.getMetadataReader(resource);
54. String className = reader.getClassMetadata().getClassName();
55. classSet.add(Class.forName(className, false, Thread.currentThread().getContextClassLoader()));
56. }
57. }
58. }catch (Exception e){
59. e.printStackTrace();
60. }
61. return classSet;
62. }
64. }

（3）在spring bean初始化后，注册RPC服务端和客户端接口，将RPC服务端相关的配置、bean、接口名称等保存到内存中，以便后续解析调用接口使用

1. /**
2. * 将rpc客户端和服务端的注解配置信息 保存到RpcFactory中
3. * @param bean
4. * @param beanName
5. */
6. public void resolveRpc(Object bean, String beanName){
7. Method[] methods = ReflectionUtils.getAllDeclaredMethods(bean.getClass());
8. if (methods == null){
9. return;
10. }
11. for (Method method : methods){
12. //解析RPC服务端接口注解的配置
13. RpcExporter rpcExporter = AnnotationUtils.findAnnotation(method, RpcExporter.class);
14. if (rpcExporter != null){
15. RpcExporterContext exporterContext = new RpcExporterContext();
16. exporterContext.setBean(bean);
17. exporterContext.setMethod(method);
18. RpcFactory.registerExporter(beanName, method.getName(), exporterContext);
19. }
20. //解析RPC客户端接口注解的配置
21. RpcClient rpcClient = AnnotationUtils.findAnnotation(method, RpcClient.class);
22. if (rpcClient != null){
23. RpcClientContext clientContext = new RpcClientContext();
24. clientContext.setUrl(rpcClient.rpcExportUrl());
25. clientContext.setMethodName(rpcClient.value());
26. RpcFactory.registerClient(beanName, method.getName(), clientContext);
27. }
28. }
29. }

（4）添加Rpc拦截器，所有rpc的请求，都通过该拦截器来执行

1. @Configuration
2. public class WebConfiguration extends WebMvcConfigurationSupport {
4. /**
5. * 配置rpc过滤器
6. * @return
7. */
8. @Bean
9. public FilterRegistrationBean rpcFilter() {
10. FilterRegistrationBean registration = new FilterRegistrationBean();
11. registration.setFilter(new RpcFilter());
12. registration.addUrlPatterns("/rpc");
13. return registration;
14. }
16. }
18. public class RpcFilter implements Filter {
20. /**
21. * 1、RPC服务端监听到特定rpc地址时，doFilter解析客户端的bean名称、接口、以及接口参数
22. * 2、调用相应的bean接口，并得到返回结果写回response
23. * @param req
24. * @param resp
25. * @param chain
26. * @throws IOException
27. * @throws ServletException
28. */
29. @Override
30. public void doFilter(ServletRequest req, ServletResponse resp, FilterChain chain) throws IOException {
31. HttpServletRequest request = (HttpServletRequest) req;
32. HttpServletResponse response = (HttpServletResponse) resp;
33. InputStreamReader inputStreamReader = null;
34. BufferedReader reader = null;
35. try{
36. inputStreamReader = new InputStreamReader(request.getInputStream(), "UTF-8");
37. reader = new BufferedReader(inputStreamReader);
38. String requestBody = "";
39. String temp ;
40. while ((temp = reader.readLine()) != null) {
41. requestBody += temp;
42. }
43. //获取请求参数
44. RpcRequest rpcRequest = JSONObject.parseObject(requestBody, RpcRequest.class);
45. if (StringUtils.isEmpty(rpcRequest.getMethod())){
46. throw new RuntimeException("rpc method parameter can not be null");
47. }
48. //获取当前rpc请求的服务端配置信息：bean和method
49. RpcExporterContext exporterContext = RpcFactory.getExporterMap(rpcRequest.getMethod());
50. //通过反射调用rpc服务端具体的接口方法
51. Object result = exporterContext.getMethod().invoke(((Class) exporterContext.getBean()).newInstance(), rpcRequest.getParams().toArray());
52. RpcResponse rpcResponse = new RpcResponse();
53. rpcResponse.setResult(result);
54. //写回执行结果
55. response.getWriter().write(JSON.toJSONString(rpcResponse));
56. }catch (Exception e){
57. e.printStackTrace();
58. }finally {
59. if (reader != null){
60. reader.close();
61. }
62. if (inputStreamReader != null){
63. inputStreamReader.close();
64. }
65. }
66. }
67. }

RPC调用链

（1）业务方调用RPC客户端接口，通过动态代理后，进入invoke方法，通过HttpClient或其他方式向RPC服务端发起请求

（2）服务端添加RPC过滤器，所有通过某个特定地址（http://xxx/rpc）来的请求，都通过这个RPC拦截器

（3）RPC拦截器中解析客户端的请求参数，包括bean名称、接口名称、接口参数等

（4）通过解析出来的请求参数，调用RPC服务端相应的接口方法，并获取到返回结果

（5）RPC服务端将请求结果返回给RPC客户端，RPC客户端接受到返回结果后，继续处理业务逻辑

添加Controller类，请求http://localhost:8080/test测试，结果返回true

1. @RestController
2. public class RpcController {
4. @Autowired
5. private RpcTestClient client;
7. @RequestMapping(value = "/test")
8. public Boolean test() throws NoSuchMethodException {
9. RpcExporterContext exporterContext = new RpcExporterContext();
10. try{
11. Class clazz = Class.forName("com.code.framework.rpc.exporter.RpcTestServiceImpl");
12. Method method = clazz.getDeclaredMethod("testRpcService", String.class);
13. exporterContext.setBean(clazz);
14. exporterContext.setMethod(method);
15. }catch (Exception e){
16. e.printStackTrace();
17. }
18. RpcFactory.registerExporter("RpcTestServiceImpl", "testRpcService", exporterContext);
20. RpcClientContext clientContext = new RpcClientContext();
21. //设置rpc服务端请求地址
22. clientContext.setUrl("http://localhost:8080/rpc");
23. //设置rpc服务端接口
24. clientContext.setMethodName("RpcTestServiceImpl.testRpcService");
25. RpcFactory.registerClient("RpcTestClient", "testClient", clientContext);
26. //上述代码为手动配置rpc客户端和服务端相应的配置信息，也可以通过扫描包中rpc客户端和服务端接口方法上的注解来获取配置信息，然后将其保存到RpcFactory中，供解析调用时使用
27. Boolean result = client.testClient("test rpc");
28. return result;
29. }
31. }