[v2.0]refractor move registry logic to request handler

README.md

 # guide-rpc-framework
+

example-client/src/main/java/github/javaguide/NettyClientMain.java

@@ -14,6 +14,7 @@ public class NettyClientMain {
         RpcClientProxy rpcClientProxy = new RpcClientProxy(rpcClient);
         HelloService helloService = rpcClientProxy.getProxy(HelloService.class);
         String hello = helloService.hello(new Hello("111", "222"));
-        System.out.println(hello);
+        //如需使用 assert 断言，需要在 VM options 添加参数：-ea
+        assert "Hello description is 222".equals(hello);
     }
 }

example-client/src/main/java/github/javaguide/RpcFrameworkSimpleClientMain.java

@@ -2,7 +2,6 @@ package github.javaguide;
 
 import github.javaguide.transport.RpcClient;
 import github.javaguide.transport.RpcClientProxy;
-import github.javaguide.transport.netty.client.NettyRpcClient;
 import github.javaguide.transport.socket.SocketRpcClient;
 
 /**

example-server/src/main/java/github/javaguide/HelloServiceImpl2.java

@@ -8,7 +8,7 @@ import org.slf4j.LoggerFactory;
  * @createTime 2020年05月12日 17:36:00
  */
 public class HelloServiceImpl2 {
-    private static final Logger logger = LoggerFactory.getLogger(HelloServiceImpl.class);
+    private static final Logger logger = LoggerFactory.getLogger(HelloServiceImpl2.class);
 
     public String hello(Hello hello) {
         logger.info("HelloServiceImpl2收到: {}.", hello.getMessage());

rpc-framework-simple/src/main/java/github/javaguide/serialize/kyro/KryoSerializer.java

@@ -45,7 +45,7 @@ public class KryoSerializer implements Serializer {
             kryoThreadLocal.remove();
             return output.toBytes();
         } catch (Exception e) {
-            logger.error("occur github.javaguide.exception when serialize:", e);
+            logger.error("occur exception when serialize:", e);
             throw new SerializeException("序列化失败");
         }
     }
@@ -60,7 +60,7 @@ public class KryoSerializer implements Serializer {
             kryoThreadLocal.remove();
             return clazz.cast(o);
         } catch (Exception e) {
-            logger.error("occur github.javaguide.exception when deserialize:", e);
+            logger.error("occur exception when deserialize:", e);
             throw new SerializeException("反序列化失败");
         }
     }

rpc-framework-simple/src/main/java/github/javaguide/transport/RpcClient.java

@@ -3,6 +3,8 @@ package github.javaguide.transport;
 import github.javaguide.dto.RpcRequest;
 
 /**
+ * 实现了 RpcClient 接口的对象需要具有发送 RpcRequest 的能力
+ *
  * @author shuang.kou
  * @createTime 2020年05月25日 17:02:00
  */

rpc-framework-simple/src/main/java/github/javaguide/transport/RpcClientProxy.java

@@ -10,22 +10,33 @@ import java.lang.reflect.Proxy;
 import java.util.UUID;
 
 /**
+ * 动态代理类。当动态代理对象调用一个方法的时候，实际调用的是下面的 invoke 方法
+ *
  * @author shuang.kou
  * @createTime 2020年05月10日 19:01:00
  */
 public class RpcClientProxy implements InvocationHandler {
     private static final Logger logger = LoggerFactory.getLogger(RpcClientProxy.class);
-    private RpcClient rpcClient;
+    /**
+     * 用于发送请求给服务端，对应socket和netty两种实现方式
+     */
+    private final RpcClient rpcClient;
 
     public RpcClientProxy(RpcClient rpcClient) {
         this.rpcClient = rpcClient;
     }
 
+    /**
+     * 通过 Proxy.newProxyInstance() 方法获取某个类的代理对象
+     */
     @SuppressWarnings("unchecked")
     public <T> T getProxy(Class<T> clazz) {
         return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
     }
 
+    /**
+     * 当你使用代理对象调用方法的时候实际会调用到这个方法。代理对象就是你通过上面的 getProxy 方法获取到的对象。
+     */
     @Override
     public Object invoke(Object proxy, Method method, Object[] args) {
         logger.info("Call invoke method and invoked method: {}", method.getName());

rpc-framework-simple/src/main/java/github/javaguide/transport/RpcRequestHandler.java

@@ -3,6 +3,8 @@ package github.javaguide.transport;
 import github.javaguide.dto.RpcRequest;
 import github.javaguide.dto.RpcResponse;
 import github.javaguide.enumeration.RpcResponseCode;
+import github.javaguide.registry.DefaultServiceRegistry;
+import github.javaguide.registry.ServiceRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -15,18 +17,31 @@ import java.lang.reflect.Method;
  */
 public class RpcRequestHandler {
     private static final Logger logger = LoggerFactory.getLogger(RpcRequestHandler.class);
+    private static final ServiceRegistry serviceRegistry;
 
-    public Object handle(RpcRequest rpcRequest, Object service) {
+    static {
+        serviceRegistry = new DefaultServiceRegistry();
+    }
+
+    /**
+     * 处理 rpcRequest 然后返回方法执行结果
+     */
+    public Object handle(RpcRequest rpcRequest) {
         Object result = null;
+        //通过注册中心获取到目标类（客户端需要调用类）
+        Object service = serviceRegistry.getService(rpcRequest.getInterfaceName());
         try {
             result = invokeTargetMethod(rpcRequest, service);
             logger.info("service:{} successful invoke method:{}", rpcRequest.getInterfaceName(), rpcRequest.getMethodName());
         } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
-            logger.error("occur github.javaguide.exception", e);
+            logger.error("occur exception", e);
         }
         return result;
     }
 
+    /**
+     * 根据 rpcRequest 和 service 对象特定的方法并返回结果
+     */
     private Object invokeTargetMethod(RpcRequest rpcRequest, Object service) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
         Method method = service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamTypes());
         if (null == method) {

rpc-framework-simple/src/main/java/github/javaguide/transport/netty/client/NettyClientHandler.java

@@ -49,7 +49,7 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
      */
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        logger.error("client catch github.javaguide.exception：", cause);
+        logger.error("client catch exception：", cause);
         cause.printStackTrace();
         ctx.close();
     }

rpc-framework-simple/src/main/java/github/javaguide/transport/netty/client/NettyRpcClient.java

@@ -81,12 +81,13 @@ public class NettyRpcClient implements RpcClient {
                 futureChannel.closeFuture().sync();
                 AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse" + rpcRequest.getRequestId());
                 RpcResponse rpcResponse = futureChannel.attr(key).get();
+                logger.info("client get rpcResponse from channel:{}", rpcResponse);
                 //校验 RpcResponse 和 RpcRequest
                 RpcMessageChecker.check(rpcResponse, rpcRequest);
                 return rpcResponse.getData();
             }
         } catch (InterruptedException e) {
-            logger.error("occur github.javaguide.exception when connect server:", e);
+            logger.error("occur exception when connect server:", e);
         }
         return null;
     }

rpc-framework-simple/src/main/java/github/javaguide/transport/netty/server/NettyRpcServer.java

@@ -27,7 +27,7 @@ import org.slf4j.LoggerFactory;
 public class NettyRpcServer {
     private static final Logger logger = LoggerFactory.getLogger(NettyRpcServer.class);
     private final int port;
-    private KryoSerializer kryoSerializer;
+    private final KryoSerializer kryoSerializer;
 
     public NettyRpcServer(int port) {
         this.port = port;
@@ -60,7 +60,7 @@ public class NettyRpcServer {
             // 等待服务端监听端口关闭
             f.channel().closeFuture().sync();
         } catch (InterruptedException e) {
-            logger.error("occur github.javaguide.exception when start server:", e);
+            logger.error("occur exception when start server:", e);
         } finally {
             bossGroup.shutdownGracefully();
             workerGroup.shutdownGracefully();

rpc-framework-simple/src/main/java/github/javaguide/transport/netty/server/NettyServerHandler.java

@@ -2,8 +2,6 @@ package github.javaguide.transport.netty.server;
 
 import github.javaguide.dto.RpcRequest;
 import github.javaguide.dto.RpcResponse;
-import github.javaguide.registry.DefaultServiceRegistry;
-import github.javaguide.registry.ServiceRegistry;
 import github.javaguide.transport.RpcRequestHandler;
 import github.javaguide.utils.concurrent.ThreadPoolFactory;
 import io.netty.channel.ChannelFuture;
@@ -29,14 +27,13 @@ import java.util.concurrent.ExecutorService;
 public class NettyServerHandler extends ChannelInboundHandlerAdapter {
 
     private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);
-    private static RpcRequestHandler rpcRequestHandler;
-    private static ServiceRegistry serviceRegistry;
-    private static ExecutorService threadPool;
+    private static final String THREAD_NAME_PREFIX = "netty-server-handler-rpc-pool";
+    private static final RpcRequestHandler rpcRequestHandler;
+    private static final ExecutorService threadPool;
 
     static {
         rpcRequestHandler = new RpcRequestHandler();
-        serviceRegistry = new DefaultServiceRegistry();
-        threadPool = ThreadPoolFactory.createDefaultThreadPool("netty-server-handler-rpc-pool");
+        threadPool = ThreadPoolFactory.createDefaultThreadPool(THREAD_NAME_PREFIX);
     }
 
     @Override
@@ -46,11 +43,8 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
             try {
                 logger.info(String.format("server receive msg: %s", msg));
                 RpcRequest rpcRequest = (RpcRequest) msg;
-                String interfaceName = rpcRequest.getInterfaceName();
-                //通过注册中心获取到目标类（客户端需要调用类）
-                Object service = serviceRegistry.getService(interfaceName);
                 //执行目标方法（客户端需要执行的方法）并且返回方法结果
-                Object result = rpcRequestHandler.handle(rpcRequest, service);
+                Object result = rpcRequestHandler.handle(rpcRequest);
                 logger.info(String.format("server get result: %s", result.toString()));
                 //返回方法执行结果给客户端
                 ChannelFuture f = ctx.writeAndFlush(RpcResponse.success(result, rpcRequest.getRequestId()));
@@ -65,7 +59,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        logger.error("server catch github.javaguide.exception");
+        logger.error("server catch exception");
         cause.printStackTrace();
         ctx.close();
     }

rpc-framework-simple/src/main/java/github/javaguide/transport/socket/SocketRpcRequestHandlerRunnable.java

@@ -2,8 +2,6 @@ package github.javaguide.transport.socket;
 
 import github.javaguide.dto.RpcRequest;
 import github.javaguide.dto.RpcResponse;
-import github.javaguide.registry.DefaultServiceRegistry;
-import github.javaguide.registry.ServiceRegistry;
 import github.javaguide.transport.RpcRequestHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,12 +18,10 @@ import java.net.Socket;
 public class SocketRpcRequestHandlerRunnable implements Runnable {
     private static final Logger logger = LoggerFactory.getLogger(SocketRpcRequestHandlerRunnable.class);
     private Socket socket;
-    private static RpcRequestHandler rpcRequestHandler;
-    private static ServiceRegistry serviceRegistry;
+    private static final RpcRequestHandler rpcRequestHandler;
 
     static {
         rpcRequestHandler = new RpcRequestHandler();
-        serviceRegistry = new DefaultServiceRegistry();
     }
 
     public SocketRpcRequestHandlerRunnable(Socket socket) {
@@ -38,13 +34,11 @@ public class SocketRpcRequestHandlerRunnable implements Runnable {
         try (ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
              ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) {
             RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();
-            String interfaceName = rpcRequest.getInterfaceName();
-            Object service = serviceRegistry.getService(interfaceName);
-            Object result = rpcRequestHandler.handle(rpcRequest, service);
+            Object result = rpcRequestHandler.handle(rpcRequest);
             objectOutputStream.writeObject(RpcResponse.success(result, rpcRequest.getRequestId()));
             objectOutputStream.flush();
         } catch (IOException | ClassNotFoundException e) {
-            logger.error("occur github.javaguide.exception:", e);
+            logger.error("occur exception:", e);
         }
     }
 

rpc-framework-simple/src/main/java/github/javaguide/transport/socket/SocketRpcServer.java

@@ -15,7 +15,7 @@ import java.util.concurrent.ExecutorService;
  */
 public class SocketRpcServer {
 
-    private ExecutorService threadPool;
+    private final ExecutorService threadPool;
     private static final Logger logger = LoggerFactory.getLogger(SocketRpcServer.class);
 
     public SocketRpcServer() {
@@ -24,7 +24,7 @@ public class SocketRpcServer {
 
     public void start(int port) {
 
-        try (ServerSocket server = new ServerSocket(port);) {
+        try (ServerSocket server = new ServerSocket(port)) {
             logger.info("server starts...");
             Socket socket;
             while ((socket = server.accept()) != null) {