devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

核心项目：MyRPC——基于Java语言实现的RPC框架_java rpc项目

作者：你好赵伟 | 2024-04-23 12:24:28

赞

踩

java rpc项目

1，项目概述

RPC：Remote Procedure Call，即远程过程调用。远程过程调用是一种像调用本地过程一样调用远程机器上的过程，而不需要了解网络细节的远程过程访问支持机制。由于基于套接字的开发方式，因此带来了开发繁琐和排错困难的问题。

项目地址：https://github.com/shao12138/MyRPC

项目架构：

项目描述：

  • 实现轻量级RPC框架，使得客户端可以通过网络从远程服务端程序上请求服务。
  • 注册中心部分使用Redis实现注册、订阅功能。
  • 在客户端实现了基于一致性哈希算法的负载均衡。
  • 动态代理部分使用JDK动态代理。
  • 网络传输部分使用Http协议进行传输。

项目流程/项目难点：

  • 理解RPC的概念和协议：要成功开发RPC，需要了解RPC的基本概念和协议。还可以使用现有的RPC框架，如gRPC或Thrift，以获得有关RPC协议的更多信息。

  • 定义服务接口：在开发RPC服务之前，需要定义RPC服务的接口。这可能涉及到定义函数的输入和输出参数，以及服务可以执行的操作。为了使服务具有可伸缩性，需要设计合理的接口，确保它们能够适应未来的需求。

  • 服务注册与发现：远程调用需要知道被调用的服务的位置，需要具备服务注册与发现的能力，如ZooKeeper、Eureka等。

  • 实现RPC客户端和服务器：在RPC客户端和服务器之间建立通信是一项重要任务。需要选择适当的通信协议（如TCP或HTTP），并编写代码以确保客户端和服务器之间的稳定通信。还需要实现序列化和反序列化机制，以便数据可以在客户端和服务器之间传输。

  • 处理并发和线程安全性：RPC服务可能会处理多个请求，因此需要确保代码能够正确地处理并发请求。还需要考虑线程安全性，以确保多个线程可以同时访问您的代码而不会导致竞态条件或死锁。

  • 安全机制：远程调用需要具备安全机制，如身份认证、授权机制等，保障数据传输的安全。

  • 负载均衡：在集群环境下，远程调用需要具备负载均衡的能力，根据不同的负载均衡策略将请求分发到不同的节点上。

2，项目实践

2.1，项目搭建

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.  
  7.     <groupId>org.nudt</groupId>
  8.     <artifactId>MyRPC</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.     <packaging>pom</packaging>
  11.     <modules>
  12.         <module>Common</module>
  13.         <module>CodeC</module>
  14.         <module>Proto</module>
  15.         <module>Transport</module>
  16.         <module>Server</module>
  17.         <module>Client</module>
  18.     </modules>
  19.  
  20.     <properties>
  21.         <maven.compiler.source>11</maven.compiler.source>
  22.         <maven.compiler.target>11</maven.compiler.target>
  23.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  24.     </properties>
  25.     <dependencyManagement>
  26.         <!--子模块可能用到的-->
  27.         <dependencies>
  28.             <dependency>
  29.                 <groupId>commons-io</groupId>
  30.                 <artifactId>commons-io</artifactId>
  31.                 <version>2.7</version>
  32.             </dependency>
  33.             <dependency>
  34.                 <groupId>org.eclipse.jetty</groupId>
  35.                 <artifactId>jetty-servlet</artifactId>
  36.                 <version>9.2.28.v20190418</version>
  37.             </dependency>
  38.             <dependency>
  39.                 <groupId>com.alibaba</groupId>
  40.                 <artifactId>fastjson</artifactId>
  41.                 <version>2.0.9</version>
  42.             </dependency>
  43.         </dependencies>
  44.     </dependencyManagement>
  45.     <dependencies>
  46.         <!--所有子模块也会被添加-->
  47.         <dependency>
  48.             <groupId>junit</groupId>
  49.             <artifactId>junit</artifactId>
  50.             <version>4.12</version>
  51.             <scope>test</scope>
  52.         </dependency>
  53.         <dependency>
  54.             <groupId>org.projectlombok</groupId>
  55.             <artifactId>lombok</artifactId>
  56.             <version>1.18.10</version>
  57.         </dependency>
  58.         <dependency>
  59.             <groupId>org.slf4j</groupId>
  60.             <artifactId>slf4j-api</artifactId>
  61.             <version>1.7.26</version>
  62.         </dependency>
  63.         <dependency>
  64.             <groupId>ch.qos.logback</groupId>
  65.             <artifactId>logback-classic</artifactId>
  66.             <version>1.2.10</version>
  67.         </dependency>
  68.     </dependencies>
  69.     <build>
  70.         <plugins>
  71.             <plugin>
  72.                 <groupId>org.apache.maven.plugins</groupId>
  73.                 <artifactId>maven-archetype-plugin</artifactId>
  74.                 <version>3.2.0</version>
  75.             </plugin>
  76.         </plugins>
  77.     </build>
  78. </project>

2.2，协议类

Peer：网络传输的一个端点

  1. package org.nudt;
  2.  
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5.  
  6. //表示网络传输的一个端点
  7. @Data   //set,get
  8. @AllArgsConstructor    //所有参数的构造器
  9. public class Peer {
  10.     //地址
  11.     private String host;
  12.     //端口号
  13.     private int port;
  14. }

ServiceDescriptor

  1. package org.nudt;
  2.  
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import java.lang.reflect.Method;
  7. import java.util.Arrays;
  8.  
  9. //表示哦服务
  10. @Data   //set,get
  11. @AllArgsConstructor    //所有参数的构造器
  12. @NoArgsConstructor     //默认构造器
  13. public class ServiceDescriptor {
  14.     //类
  15.     private String clazz;
  16.     //方法
  17.     private String method;
  18.     //返回类型
  19.     private String returnType;
  20.     //参数类型
  21.     private String[] parameterTypes;
  22.  
  23.     public static ServiceDescriptor from(Class clazz, Method method) {
  24.         ServiceDescriptor sdp = new ServiceDescriptor();
  25.         sdp.setClazz(clazz.getName());
  26.         sdp.setMethod(method.getName());
  27.         sdp.setReturnType(method.getReturnType().getTypeName());
  28.  
  29.         Class[] parameterClasses = method.getParameterTypes();
  30.         String[] parameterTypes = new String[parameterClasses.length];
  31.         for (int i = 0; i < parameterClasses.length; i++) {
  32.             parameterTypes[i] = parameterClasses[i].getName();
  33.         }
  34.         sdp.setParameterTypes(parameterTypes);
  35.         return sdp;
  36.     }
  37.  
  38.     public boolean equals(Object o) {
  39.         if (this == o) {
  40.             return true;
  41.         }
  42.         if (o == null || getClass() != o.getClass()) {
  43.             return false;
  44.         }
  45.         ServiceDescriptor that = (ServiceDescriptor) o;
  46.         return this.toString().equals(that.toString());
  47.     }
  48.  
  49.     public int hashCode() {
  50.         return toString().hashCode();
  51.     }
  52.  
  53.     public String toString() {
  54.         return "ServiceDescriptor{" +
  55.                 "clazz='" + clazz + '\'' +
  56.                 ", method='" + method + '\'' +
  57.                 ", returnType='" + returnType + '\'' +
  58.                 ", parameterTypes=" + Arrays.toString(parameterTypes) +
  59.                 '}';
  60.     }
  61. }

Request&Response

  1. package org.nudt;
  2.  
  3. import lombok.Data;
  4.  
  5. @Data   //set,get
  6. public class Request {
  7.     //请求服务
  8.     private ServiceDescriptor service;
  9.     //请求参数
  10.     private Object[] parameter;
  11. }
  1. package org.nudt;
  2.  
  3. import lombok.Data;
  4.  
  5. //表示RPC的响应
  6. @Data   //set,get
  7. public class Response {
  8.     //服务返回信息：0成功，非0失败
  9.     private int code = 0;
  10.     //具体错误信息
  11.     private String message = "ok";
  12.     //返回数据
  13.     private Object data;
  14. }

2.3，反射工具类

ReflectionUtils

  1. package org.nudt;
  2.  
  3. import java.lang.reflect.Method;
  4. import java.lang.reflect.Modifier;
  5. import java.util.ArrayList;
  6. import java.util.List;
  7.  
  8. //反射工具类
  9. public class ReflectionUtils {
  10.     //根据Class创建对象
  11.     public static <T> T newInstance(Class clazz) {
  12.         try {
  13.             return (T) clazz.newInstance();
  14.         } catch (Exception e) {
  15.             throw new IllegalStateException(e);
  16.         }
  17.     }
  18.  
  19.     //获取某个类所有的公共方法
  20.     public static Method[] getPublicMethods(Class clazz) {
  21.         Method[] methods = clazz.getDeclaredMethods();
  22.         List<Method> pmethods = new ArrayList<>();
  23.         for (Method m : methods) {
  24.             if (Modifier.isPublic(m.getModifiers())) {
  25.                 pmethods.add(m);
  26.             }
  27.         }
  28.         return pmethods.toArray(new Method[0]);
  29.     }
  30.  
  31.     //调用某个对象的指定方法
  32.     public static Object invoke(Object obj, Method method, Object... args) {
  33.         try {
  34.             return method.invoke(obj, args);
  35.         } catch (Exception e) {
  36.             throw new IllegalStateException(e);
  37.         }
  38.     }
  39. }

2.4，序列化模块

引入fastjson

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>com.alibaba</groupId>
  4.         <artifactId>fastjson</artifactId>
  5.         <version>2.0.9</version>
  6.     </dependency>
  7. </dependencies>

Decoder&Encoder

  1. package org.nudt;
  2.  
  3. //反序列化
  4. public interface Decoder {
  5.     <T> T decode(byte[] bytes, Class<T> clazz);
  6. }
  1. package org.nudt;
  2.  
  3. //序列化
  4. public interface Encoder {
  5.     byte[] encode(Object obj);
  6. }

JSONEncoder&JSONDecoder

  1. package org.nudt;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4.  
  5. //基于JSON的序列化实现
  6. public class JSONEncoder implements Encoder {
  7.     public byte[] encode(Object obj) {
  8.         return JSON.toJSONBytes(obj);
  9.     }
  10. }
  1. package org.nudt;
  2.  
  3. import com.alibaba.fastjson2.JSON;
  4.  
  5. //基于JSON的反序列化
  6. public class JSONDecoder implements Decoder {
  7.     public <T> T decode(byte[] bytes, Class<T> clazz) {
  8.         return JSON.parseObject(bytes, clazz);
  9.     }
  10. }

3，网络模块

3.1，接口

引入依赖

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>commons-io</groupId>
  4.         <artifactId>commons-io</artifactId>
  5.         <version>2.7</version>
  6.     </dependency>
  7.     <dependency>
  8.         <groupId>org.eclipse.jetty</groupId>
  9.         <artifactId>jetty-servlet</artifactId>
  10.         <version>9.2.28.v20190418</version>
  11.     </dependency>
  12.     <dependency>
  13.         <groupId>org.nudt</groupId>
  14.         <artifactId>Proto</artifactId>
  15.         <version>1.0-SNAPSHOT</version>
  16.     </dependency>
  17. </dependencies>

TransportClient&TranspotServer

  1. package org.nudt;
  2.  
  3. import java.io.InputStream;
  4.  
  5. /*
  6.  * 创建连接
  7.  * 发送数据，并且等待响应
  8.  * 关闭连接*/
  9. public interface TransportClient {
  10.     void connect(Peer peer);
  11.  
  12.     InputStream write(InputStream data);
  13.  
  14.     void close();
  15. }
  1. package org.nudt;
  2.  
  3. /*启动，监听
  4.  * 接受请求
  5.  * 关闭监听*/
  6. public interface TransportServer {
  7.     void init(int prot, RequestHandler handler);
  8.  
  9.     void start();
  10.  
  11.     void stop();
  12. }

RequestHandler

  1. package org.nudt;
  2.  
  3. import java.io.InputStream;
  4. import java.io.OutputStream;
  5.  
  6. //处理网络请求的handler
  7. public interface RequestHandler {
  8.     void onRequest(InputStream revice, OutputStream toResponse);
  9. }

3.2，实现Client

HttpTransportClient

  1. package org.nudt;
  2.  
  3. import org.apache.commons.io.IOUtils;
  4.  
  5. import java.io.IOException;
  6. import java.io.InputStream;
  7. import java.net.HttpURLConnection;
  8. import java.net.URL;
  9.  
  10. public class HTTPTransportClient implements TransportClient {
  11.     private String url;
  12.  
  13.     public void connect(Peer peer) {
  14.         this.url = "http://" + peer.getHost() + ":" + peer.getPort();
  15.  
  16.     }
  17.  
  18.     public InputStream write(InputStream data) {
  19.         try {
  20.             HttpURLConnection httpConn = (HttpURLConnection) new URL(url).openConnection();
  21.             httpConn.setDoInput(true);
  22.             httpConn.setDoOutput(true);
  23.             httpConn.setUseCaches(false);
  24.             httpConn.setRequestMethod("POST");
  25.             httpConn.connect();
  26.  
  27.             IOUtils.copy(data, httpConn.getOutputStream());
  28.  
  29.             int resultCode = httpConn.getResponseCode();
  30.             if (resultCode == HttpURLConnection.HTTP_OK) {
  31.                 return httpConn.getInputStream();
  32.             } else {
  33.                 return httpConn.getErrorStream();
  34.             }
  35.         } catch (IOException e) {
  36.             throw new IllegalStateException(e);
  37.         }
  38.     }
  39.  
  40.     public void close() {
  41.  
  42.     }
  43. }

3.3，实现Server

HttpTransportServer

  1. package org.nudt;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.eclipse.jetty.server.Server;
  5. import org.eclipse.jetty.servlet.ServletContextHandler;
  6. import org.eclipse.jetty.servlet.ServletHolder;
  7.  
  8. import javax.servlet.ServletException;
  9. import javax.servlet.http.HttpServlet;
  10. import javax.servlet.http.HttpServletRequest;
  11. import javax.servlet.http.HttpServletResponse;
  12. import java.io.IOException;
  13. import java.io.InputStream;
  14. import java.io.OutputStream;
  15.  
  16. @Slf4j
  17. public class HttpTransportServer implements TransportServer {
  18.     private RequestHandler handler;
  19.     private Server server;
  20.  
  21.     public void init(int prot, RequestHandler handler) {
  22.         this.handler = handler;
  23.         this.server = new Server(prot);
  24.         //Servlet接收请求
  25.         ServletContextHandler ctx = new ServletContextHandler();
  26.         server.setHandler(ctx);
  27.  
  28.         ServletHolder holder = new ServletHolder(new RequestServlet());
  29.         ctx.addServlet(holder, "/*");
  30.  
  31.     }
  32.  
  33.     public void start() {
  34.         try {
  35.             server.start();
  36.             server.join();
  37.         } catch (Exception e) {
  38.             throw new RuntimeException(e);
  39.         }
  40.     }
  41.  
  42.     public void stop() {
  43.         try {
  44.             server.start();
  45.         } catch (Exception e) {
  46.             throw new RuntimeException(e);
  47.         }
  48.     }
  49.  
  50.     class RequestServlet extends HttpServlet {
  51.         protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
  52.             log.info("client connect");
  53.             InputStream in = req.getInputStream();
  54.             OutputStream out = resp.getOutputStream();
  55.             if (handler != null) {
  56.                 handler.onRequest(in, out);
  57.             }
  58.             out.flush();
  59.         }
  60.     }
  61. }

4，Server模块

4.1，配置

引入依赖

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.nudt</groupId>
  4.         <artifactId>Proto</artifactId>
  5.         <version>1.0-SNAPSHOT</version>
  6.     </dependency>
  7.     <dependency>
  8.         <groupId>org.nudt</groupId>
  9.         <artifactId>CodeC</artifactId>
  10.         <version>1.0-SNAPSHOT</version>
  11.     </dependency>
  12.     <dependency>
  13.         <groupId>org.nudt</groupId>
  14.         <artifactId>Common</artifactId>
  15.         <version>1.0-SNAPSHOT</version>
  16.      </dependency>
  17.      <dependency>
  18.         <groupId>commons-io</groupId>
  19.         <artifactId>commons-io</artifactId>
  20.      </dependency>
  21.      <dependency>
  22.         <groupId>org.nudt</groupId>
  23.         <artifactId>Transport</artifactId>
  24.         <version>1.0-SNAPSHOT</version>
  25.         <scope>compile</scope>
  26.      </dependency>
  27. </dependencies>

RpcServerConfig

  1. package org.nudt;
  2.  
  3. import lombok.Data;
  4.  
  5. //server配置
  6. @Data
  7. public class RpcServerConfig {
  8.     private Class<? extends TransportServer> transportClass = HttpTransportServer.class;
  9.     private Class<? extends Encoder> encoderClass = JSONEncoder.class;
  10.     private Class<? extends Decoder> decoderClass = JSONDecoder.class;
  11.     private int port = 3000;
  12.  
  13. }

RpcServer

  1. package org.nudt;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.commons.io.IOUtils;
  5.  
  6. import java.io.IOException;
  7. import java.io.InputStream;
  8. import java.io.OutputStream;
  9.  
  10. @Slf4j
  11. public class RpcServer {
  12.     private RpcServerConfig config;
  13.     private TransportServer net;
  14.     private Encoder encoder;
  15.     private Decoder decoder;
  16.     private ServiceManager serviceManager;
  17.     private ServiceInvoker serviceInvoker;
  18.  
  19.     public RpcServer(RpcServerConfig config) {
  20.         this.config = config;
  21.         //net
  22.         this.net = ReflectionUtils.newInstance(config.getTransportClass());
  23.         this.net.init(config.getPort(), this.handler);
  24.         //序列化
  25.         this.encoder = ReflectionUtils.newInstance(config.getEncoderClass());
  26.         //反序列化
  27.         this.decoder = ReflectionUtils.newInstance(config.getDecoderClass());
  28.         //service
  29.         this.serviceManager = new ServiceManager();
  30.         this.serviceInvoker = new ServiceInvoker();
  31.     }
  32.  
  33.     public <T> void register(Class<T> interfaceClass, T bean) {
  34.         serviceManager.register(interfaceClass, bean);
  35.     }
  36.  
  37.     public void start() {
  38.         this.net.start();
  39.     }
  40.  
  41.     public void stop() {
  42.         this.net.stop();
  43.     }
  44.  
  45.     private RequestHandler handler = new RequestHandler() {
  46.         public void onRequest(InputStream revice, OutputStream toResponse) {
  47.             Response resp = new Response();
  48.             try {
  49.                 byte[] inBytes = IOUtils.readFully(revice, revice.available());
  50.                 Request request = decoder.decode(inBytes, Request.class);
  51.                 log.info("get request:{}", request);
  52.                 ServiceInstance sis = serviceManager.lookup(request);
  53.                 Object ret = serviceInvoker.invoke(sis, request);
  54.                 resp.setData(ret);
  55.             } catch (Exception e) {
  56.                 resp.setData(1);
  57.                 resp.setMessage("RpcServer got error:" + e.getClass() + " : " + e.getMessage());
  58.             } finally {
  59.                 byte[] outBytes = encoder.encode(resp);
  60.                 try {
  61.                     toResponse.write(outBytes);
  62.                 } catch (IOException e) {
  63.                     log.warn(e.getMessage(), e);
  64.                 }
  65.             }
  66.  
  67.         }
  68.     };
  69. }

4.2，Server

ServiceInstance

  1. package org.nudt;
  2.  
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5.  
  6. import java.lang.reflect.Method;
  7.  
  8. //表示一个具体服务
  9. @Data
  10. @AllArgsConstructor
  11. public class ServiceInstance {
  12.     private Object target;
  13.     private Method method;
  14.  
  15. }

ServiceInvoker

  1. package org.nudt;
  2. //调用具体服务
  3. public class ServiceInvoker {
  4.     public Object invoke(ServiceInstance service, Request request) {
  5.         return ReflectionUtils.invoke(service.getTarget(),service.getMethod(),request.getParameter());
  6.     }
  7. }

ServiceManager

  1. package org.nudt;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4.  
  5. import java.lang.reflect.Method;
  6. import java.util.Map;
  7. import java.util.concurrent.ConcurrentHashMap;
  8.  
  9. //管理RPC暴露的服务
  10. @Slf4j
  11. public class ServiceManager {
  12.     private Map<ServiceDescriptor, ServiceInstance> services;
  13.  
  14.     public ServiceManager() {
  15.         this.services = new ConcurrentHashMap<>();
  16.     }
  17.  
  18.     //注册：扫描接口所有方法
  19.     public <T> void register(Class<T> interfaceClass, T bean) {
  20.         Method[] methods = ReflectionUtils.getPublicMethods(interfaceClass);
  21.         for (Method method : methods) {
  22.             ServiceInstance sis = new ServiceInstance(bean, method);
  23.             ServiceDescriptor sdp = ServiceDescriptor.from(interfaceClass, method);
  24.             services.put(sdp, sis);
  25.             log.info("有服务被注册了:{}  {}", sdp.getClazz(), sdp.getMethod());
  26.         }
  27.     }
  28.     //查找服务
  29.     public ServiceInstance lookup(Request request){
  30.         ServiceDescriptor sdp = request.getService();
  31.         return services.get(sdp);
  32.     }
  33. }

5，Client模块

5.1，配置

引入配置

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.nudt</groupId>
  4.         <artifactId>Proto</artifactId>
  5.         <version>1.0-SNAPSHOT</version>
  6.     </dependency>
  7.     <dependency>
  8.         <groupId>org.nudt</groupId>
  9.         <artifactId>CodeC</artifactId>
  10.         <version>1.0-SNAPSHOT</version>
  11.     </dependency>
  12.     <dependency>
  13.         <groupId>org.nudt</groupId>
  14.         <artifactId>Common</artifactId>
  15.         <version>1.0-SNAPSHOT</version>
  16.     </dependency>
  17.     <dependency>
  18.         <groupId>org.nudt</groupId>
  19.         <artifactId>Transport</artifactId>
  20.         <version>1.0-SNAPSHOT</version>
  21.     </dependency>
  22.     <dependency>
  23.         <groupId>commons-io</groupId>
  24.         <artifactId>commons-io</artifactId>
  25.     </dependency>
  26. </dependencies>

RpcClientConfig

  1. package org.nudt;
  2.  
  3. import lombok.Data;
  4.  
  5. import java.util.Arrays;
  6. import java.util.List;
  7.  
  8. @Data
  9. public class RpcClientConfig {
  10.     private Class<? extends TransportClient> transportClass = HTTPTransportClient.class;
  11.     private Class<? extends Encoder> encoderClass = JSONEncoder.class;
  12.     private Class<? extends Decoder> decoderClass = JSONDecoder.class;
  13.     private Class<? extends TransportSelector> selectorClass = RandomTransportSelector.class;
  14.     private int connectCount = 1;
  15.     private List<Peer> servers = Arrays.asList(new Peer("127.0.0.1", 3000));
  16. }

TransportSelector

  1. package org.nudt;
  2.  
  3. import java.util.List;
  4.  
  5. //选择哪个Server去连接
  6. public interface TransportSelector {
  7.     /*初始化
  8.      * peers：可以连接的server端点信息
  9.      * count：client与server建立多少个连接
  10.      * clazz：client实现class*/
  11.     void init(List<Peer> peers, int count, Class<? extends TransportClient> clazz);
  12.  
  13.     //选择一个transport与Server做交互
  14.     TransportClient select();
  15.  
  16.     //释放client
  17.     void release(TransportClient client);
  18.  
  19.     //关闭client
  20.     void close();
  21. }

RandomTransportSelector

  1. package org.nudt;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4.  
  5. import java.util.ArrayList;
  6. import java.util.List;
  7. import java.util.Random;
  8.  
  9. @Slf4j
  10. public class RandomTransportSelector implements TransportSelector {
  11.     //已经连接好的client
  12.     private List<TransportClient> clients;
  13.  
  14.     public RandomTransportSelector() {
  15.         clients = new ArrayList<>();
  16.     }
  17.  
  18.     public synchronized void init(List<Peer> peers, int count, Class<? extends TransportClient> clazz) {
  19.         count = Math.max(count, 1);
  20.         for (Peer peer : peers) {
  21.             for (int i = 0; i < count; i++) {
  22.                 TransportClient client = ReflectionUtils.newInstance(clazz);
  23.                 client.connect(peer);
  24.                 clients.add(client);
  25.             }
  26.             log.info("connect server:{}", peer);
  27.         }
  28.     }
  29.  
  30.     public synchronized TransportClient select() {
  31.         int i = new Random().nextInt(clients.size());
  32.         return clients.remove(i);
  33.     }
  34.  
  35.     public synchronized void release(TransportClient client) {
  36.         clients.add(client);
  37.     }
  38.  
  39.     public synchronized void close() {
  40.         for (TransportClient cliet : clients) {
  41.             cliet.close();
  42.         }
  43.         clients.clear();
  44.     }
  45. }

5.2，Client

RpcClient

  1. package org.nudt;
  2.  
  3. import java.lang.reflect.Proxy;
  4.  
  5. public class RpcClient {
  6.     private RpcClientConfig config;
  7.     private Encoder encoder;
  8.     private Decoder decoder;
  9.     private TransportSelector selector;
  10.  
  11.     public RpcClient() {
  12.         this(new RpcClientConfig());
  13.     }
  14.  
  15.     public RpcClient(RpcClientConfig config) {
  16.         this.config = config;
  17.         this.encoder = ReflectionUtils.newInstance(this.config.getEncoderClass());
  18.         this.decoder = ReflectionUtils.newInstance(this.config.getDecoderClass());
  19.         this.selector = ReflectionUtils.newInstance(this.config.getSelectorClass());
  20.  
  21.         this.selector.init(this.config.getServers(), this.config.getConnectCount(), this.config.getTransportClass());
  22.     }
  23.  
  24.     //获取接口代理对象
  25.     public <T> T getProxy(Class<T> clazz) {
  26.         return (T) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{clazz}, new RemoteInvoker(clazz, encoder, decoder, selector));
  27.     }
  28.  
  29. }

RemoteInvoker

  1. package org.nudt;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.commons.io.IOUtils;
  5.  
  6. import java.io.ByteArrayInputStream;
  7. import java.io.InputStream;
  8. import java.lang.reflect.InvocationHandler;
  9. import java.lang.reflect.Method;
  10.  
  11. @Slf4j
  12. //调用远程服务代理类
  13. public class RemoteInvoker implements InvocationHandler {
  14.     private Class clazz;
  15.     private Encoder encoder;
  16.     private Decoder decoder;
  17.     private TransportSelector selector;
  18.  
  19.     public RemoteInvoker(Class clazz, Encoder encoder, Decoder decoder, TransportSelector selector) {
  20.         this.clazz = clazz;
  21.         this.encoder = encoder;
  22.         this.decoder = decoder;
  23.         this.selector = selector;
  24.     }
  25.  
  26.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  27.         Request request = new Request();
  28.         request.setService(ServiceDescriptor.from(clazz, method));
  29.         request.setParameter(args);
  30.         Response response = invokeRemote(request);
  31.         if (response == null || response.getCode() != 0) {
  32.             throw new IllegalStateException("fail to invoke remote:" + response);
  33.         }
  34.         return response.getData();
  35.     }
  36.  
  37.     private Response invokeRemote(Request request) {
  38.         Response resp = null;
  39.         TransportClient client = null;
  40.         try {
  41.             client = selector.select();
  42.             byte[] outBytes = encoder.encode(request);
  43.             InputStream revice = client.write(new ByteArrayInputStream(outBytes));
  44.             byte[] inBytes = IOUtils.readFully(revice, revice.available());
  45.             resp = decoder.decode(inBytes, Response.class);
  46.         } catch (Exception e) {
  47.             log.warn(e.getMessage(), e);
  48.             resp = new Response();
  49.             resp.setCode(1);
  50.             resp.setMessage("RpcClient got error:" + e.getClass() + ":" + e.getMessage());
  51.  
  52.         } finally {
  53.             if (client != null) {
  54.                 selector.release(client);
  55.             }
  56.         }
  57.         return resp;
  58.     }
  59. }

6，测试

CalcService

  1. package org.nudt;
  2.  
  3. public interface CalcService {
  4.     int add(int a, int b);
  5.  
  6.     int minus(int a, int b);
  7. }

CalcServiceImpl

  1. package org.nudt;
  2.  
  3. public class CalcServiceImpl implements CalcService {
  4.     public int add(int a, int b) {
  5.         return a + b;
  6.     }
  7.  
  8.     @Override
  9.     public int minus(int a, int b) {
  10.         return a - b;
  11.     }
  12. }

Client

  1. package org.nudt;
  2.  
  3. public class Client {
  4.     public static void main(String[] args) {
  5.         RpcClient client = new RpcClient();
  6.         CalcService service = client.getProxy(CalcService.class);
  7.         int r1 = service.add(1, 2);
  8.         int r2 = service.minus(1, 2);
  9.         System.out.println(r1 + " " + r2);
  10.     }
  11. }

Server

  1. package org.nudt;
  2.  
  3. public class Server {
  4.     public static void main(String[] args) {
  5.         RpcServer server = new RpcServer(new RpcServerConfig());
  6.         server.register(CalcService.class, new CalcServiceImpl());
  7.         server.start();
  8.     }
  9. }
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/473861
推荐阅读
相关标签

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

  

闽ICP备14008679号