引入注册表，将反射调用与线程池解耦

rpc-common/src/main/java/cn/fzzfrjf/entity/RpcResponse.java

@@ -14,15 +14,15 @@ public class RpcResponse<T> implements Serializable {
 
     private T data;
 
-    public <T> RpcResponse<T> success(T data,String requestId){
+    public static <T> RpcResponse<T> success(T data,String requestId){
         RpcResponse<T> rpcResponse = new RpcResponse<>();
         rpcResponse.setCode(ResponseCode.SUCCESS.getCode());
         rpcResponse.setRequestId(requestId);
         rpcResponse.setData(data);
         return rpcResponse;
     }
 
-    public  <T>RpcResponse<T> fail(String requestId){
+    public static <T>RpcResponse<T> fail(String requestId){
         RpcResponse<T> rpcResponse = new RpcResponse();
         rpcResponse.setRequestId(requestId);
         rpcResponse.setCode(ResponseCode.FAILURE.getCode());

rpc-common/src/main/java/cn/fzzfrjf/enumeration/RpcError.java

@@ -0,0 +1,15 @@
+package cn.fzzfrjf.enumeration;
+
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public enum RpcError {
+
+    SERVICE_NOT_REGISTERED("服务未能成功注册"),
+    SERVICE_NOT_FOUND("未能找到服务");
+
+    private String message;
+}

rpc-common/src/main/java/cn/fzzfrjf/exception/RpcException.java

@@ -0,0 +1,18 @@
+package cn.fzzfrjf.exception;
+
+import cn.fzzfrjf.enumeration.RpcError;
+
+public class RpcException extends RuntimeException{
+
+    public RpcException(RpcError rpcError,String detail){
+        super(rpcError.getMessage() + detail);
+    }
+
+    public RpcException(RpcError rpcError){
+        super(rpcError.getMessage());
+    }
+
+    public RpcException(RpcError rpcError,Throwable cause){
+        super(rpcError.getMessage(),cause);
+    }
+}

rpc-server/src/main/java/cn/fzzfrjf/core/CommonServer.java

@@ -1,5 +1,5 @@
 package cn.fzzfrjf.core;
 
 public interface CommonServer {
-    public void register(Object service ,int port);
+    public void start(int port);
 }

rpc-server/src/main/java/cn/fzzfrjf/core/DefaultServerPublisher.java

@@ -0,0 +1,41 @@
+package cn.fzzfrjf.core;
+
+import cn.fzzfrjf.enumeration.RpcError;
+import cn.fzzfrjf.exception.RpcException;
+import cn.fzzfrjf.service.ServerPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class DefaultServerPublisher implements ServerPublisher {
+
+    private static final Logger logger = LoggerFactory.getLogger(DefaultServerPublisher.class);
+    private final ConcurrentHashMap<String,Object> serviceMap = new ConcurrentHashMap<>();
+    private final Set<String> registeredService = ConcurrentHashMap.newKeySet();
+    @Override
+    public synchronized <T> void publishService(T service) {
+        String serviceName = service.getClass().getCanonicalName();
+        if(serviceMap.containsKey(serviceName)) return;
+        registeredService.add(serviceName);
+        Class<?>[] interfaces = service.getClass().getInterfaces();
+        if(interfaces.length == 0){
+            throw new RpcException(RpcError.SERVICE_NOT_REGISTERED);
+        }
+        for (Class<?> anInterface : interfaces) {
+            serviceMap.put(anInterface.getCanonicalName(),service);
+        }
+        logger.info("向接口：{}注册服务：{}",interfaces,serviceName);
+    }
+
+    @Override
+    public synchronized Object getService(String serviceName) {
+        Object o = serviceMap.get(serviceName);
+        if(null == o){
+            throw new RpcException(RpcError.SERVICE_NOT_FOUND);
+        }
+        return o;
+    }
+}

rpc-server/src/main/java/cn/fzzfrjf/core/RequestHandler.java

@@ -0,0 +1,35 @@
+package cn.fzzfrjf.core;
+
+import cn.fzzfrjf.entity.RpcRequest;
+import cn.fzzfrjf.entity.RpcResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+
+public class RequestHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(RequestHandler.class);
+
+    public Object handle(RpcRequest rpcRequest,Object service){
+        Object result = null;
+        try{
+            result = doMethod(rpcRequest,service);
+        }catch (InvocationTargetException|IllegalAccessException e){
+            logger.error("反射调用时发生错误：",e);
+        }
+        return result;
+    }
+
+    private Object doMethod(RpcRequest rpcRequest , Object service) throws InvocationTargetException, IllegalAccessException {
+        Method method;
+        try{
+            method = service.getClass().getMethod(rpcRequest.getMethodName(),rpcRequest.getParameterTypes());
+        }catch (NoSuchMethodException e){
+            logger.error("调用方法时有错误发生：",e);
+            return RpcResponse.fail(rpcRequest.getRequestId());
+        }
+        return method.invoke(service,rpcRequest.getParameters());
+    }
+}

rpc-server/src/main/java/cn/fzzfrjf/core/RequestHandlerThread.java

@@ -0,0 +1,41 @@
+package cn.fzzfrjf.core;
+
+import cn.fzzfrjf.entity.RpcRequest;
+import cn.fzzfrjf.entity.RpcResponse;
+import cn.fzzfrjf.service.ServerPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.net.Socket;
+
+public class RequestHandlerThread implements Runnable{
+
+    private static final Logger logger = LoggerFactory.getLogger(RequestHandlerThread.class);
+
+    private final RequestHandler handler;
+    private final ServerPublisher serverPublisher;
+    private final Socket socket;
+
+    public RequestHandlerThread(RequestHandler handler,ServerPublisher serverPublisher,Socket socket){
+        this.handler = handler;
+        this.serverPublisher = serverPublisher;
+        this.socket = socket;
+    }
+
+    @Override
+    public void run() {
+        try(ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
+            ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream())){
+            RpcRequest rpcRequest = (RpcRequest) ois.readObject();
+            Object service = serverPublisher.getService(rpcRequest.getInterfaceName());
+            Object result = handler.handle(rpcRequest, service);
+            oos.writeObject(RpcResponse.success(result,rpcRequest.getRequestId()));
+            oos.flush();
+        }catch (IOException | ClassNotFoundException e){
+            logger.error("调用或发送时发生错误：",e);
+        }
+    }
+}

rpc-server/src/main/java/cn/fzzfrjf/core/SocketServer.java

@@ -1,6 +1,7 @@
 package cn.fzzfrjf.core;
 
 
+import cn.fzzfrjf.service.ServerPublisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -13,26 +14,31 @@ public class SocketServer implements CommonServer{
 
     private final ExecutorService threadPool;
     private static final Logger logger = LoggerFactory.getLogger(SocketServer.class);
+    private final ServerPublisher serverPublisher;
 
-    public SocketServer(){
+
+    public SocketServer(ServerPublisher serverPublisher){
+        this.serverPublisher = serverPublisher;
         int corePoolSize = 5;
         int maximumPoolSize = 50;
         int keepAliveTime = 60;
         BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(100);
         threadPool = new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime, TimeUnit.SECONDS,workQueue,Executors.defaultThreadFactory());
     }
 
+
     @Override
-    public void register(Object service, int port) {
+    public void start(int port){
         try(ServerSocket serverSocket = new ServerSocket(port)){
-            logger.info("服务器正在启动中.....");
-            Socket socket ;
-            while( (socket = serverSocket.accept()) != null){
+            logger.info("服务器成功启动。。。。");
+            Socket socket;
+            while((socket = serverSocket.accept()) != null){
                 logger.info("连接成功，客户端ip为:" + socket.getInetAddress());
-                threadPool.execute(new WorkThread(service,socket));
+                threadPool.execute(new RequestHandlerThread(new RequestHandler(),serverPublisher,socket));
             }
+            threadPool.shutdown();
         }catch (IOException e){
-            logger.error("连接时有错误发生：{}",e);
+            logger.error("服务器启动时发生错误：",e);
         }
     }
 }

rpc-server/src/main/java/cn/fzzfrjf/core/WorkThread.java

@@ -29,7 +29,7 @@ public void run() {
             RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();
             Method method = service.getClass().getMethod(rpcRequest.getMethodName(),rpcRequest.getParameterTypes());
             Object result = method.invoke(service,rpcRequest.getParameters());
-            objectOutputStream.writeObject(new RpcResponse().success(result,rpcRequest.getRequestId()));
+            objectOutputStream.writeObject(RpcResponse.success(result,rpcRequest.getRequestId()));
             objectOutputStream.flush();
         }catch (IOException | ClassNotFoundException | InvocationTargetException |NoSuchMethodException | IllegalAccessException e){
             logger.error("调用或发送时有错误发生：",e);

rpc-server/src/main/java/cn/fzzfrjf/service/ServerPublisher.java

@@ -0,0 +1,8 @@
+package cn.fzzfrjf.service;
+
+public interface ServerPublisher {
+
+    <T> void publishService(T service);
+
+    Object getService(String serviceName);
+}

test-client/src/main/java/cn/fzzfrjf/test/SocketClientTest.java

@@ -10,7 +10,7 @@ public static void main(String[] args) {
         SocketClient socketClient = new SocketClient();
         ClientProxy proxy = new ClientProxy(socketClient,"127.0.0.1",9000);
         HelloService service = (HelloService)proxy.getProxy(HelloService.class);
-        RpcObject rpcObject = new RpcObject(1,"This is SocketClient!");
+        RpcObject rpcObject = new RpcObject(2,"This is SocketClient!");
         String s = service.sayHello(rpcObject);
         System.out.println(s);
     }

test-server/src/main/java/cn/fzzfrjf/test/SocketServerTest.java

@@ -1,13 +1,17 @@
 package cn.fzzfrjf.test;
 
+import cn.fzzfrjf.core.DefaultServerPublisher;
 import cn.fzzfrjf.core.SocketServer;
 import cn.fzzfrjf.entity.HelloService;
 import cn.fzzfrjf.service.HelloServiceImpl;
+import cn.fzzfrjf.service.ServerPublisher;
 
 public class SocketServerTest {
     public static void main(String[] args) {
-        SocketServer socketServer = new SocketServer();
+        ServerPublisher serverPublisher = new DefaultServerPublisher();
         HelloService helloService = new HelloServiceImpl();
-        socketServer.register(helloService,9000);
+        serverPublisher.publishService(helloService);
+        SocketServer socketServer = new SocketServer(serverPublisher);
+        socketServer.start(9000);
     }
 }