Merge pull request #41 from mesmerizeBy/master

Snailclimb · web-flow · commit 336316ceb87b · 2021-04-28T10:45:52.000+08:00
Fix consistency hash bug
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/AbstractLoadBalance.java b/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/AbstractLoadBalance.java
@@ -1,5 +1,7 @@
 package github.javaguide.loadbalance;
 
+import github.javaguide.remoting.dto.RpcRequest;
+
 import java.util.List;
 
 /**
@@ -10,16 +12,16 @@
  */
 public abstract class AbstractLoadBalance implements LoadBalance {
     @Override
-    public String selectServiceAddress(List<String> serviceAddresses, String rpcServiceName) {
+    public String selectServiceAddress(List<String> serviceAddresses, RpcRequest rpcRequest) {
         if (serviceAddresses == null || serviceAddresses.size() == 0) {
             return null;
         }
         if (serviceAddresses.size() == 1) {
             return serviceAddresses.get(0);
         }
-        return doSelect(serviceAddresses, rpcServiceName);
+        return doSelect(serviceAddresses, rpcRequest);
     }
 
-    protected abstract String doSelect(List<String> serviceAddresses, String rpcServiceName);
+    protected abstract String doSelect(List<String> serviceAddresses, RpcRequest rpcRequest);
 
 }
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/LoadBalance.java b/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/LoadBalance.java
@@ -1,6 +1,7 @@
 package github.javaguide.loadbalance;
 
 import github.javaguide.extension.SPI;
+import github.javaguide.remoting.dto.RpcRequest;
 
 import java.util.List;
 
@@ -18,5 +19,5 @@ public interface LoadBalance {
      * @param serviceAddresses Service address list
      * @return target service address
      */
-    String selectServiceAddress(List<String> serviceAddresses, String rpcServiceName);
+    String selectServiceAddress(List<String> serviceAddresses, RpcRequest rpcRequest);
 }
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/loadbalancer/ConsistentHashLoadBalance.java b/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/loadbalancer/ConsistentHashLoadBalance.java
@@ -1,11 +1,13 @@
 package github.javaguide.loadbalance.loadbalancer;
 
 import github.javaguide.loadbalance.AbstractLoadBalance;
+import github.javaguide.remoting.dto.RpcRequest;
 import lombok.extern.slf4j.Slf4j;
 
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
@@ -22,8 +24,10 @@ public class ConsistentHashLoadBalance extends AbstractLoadBalance {
     private final ConcurrentHashMap<String, ConsistentHashSelector> selectors = new ConcurrentHashMap<>();
 
     @Override
-    protected String doSelect(List<String> serviceAddresses, String rpcServiceName) {
+    protected String doSelect(List<String> serviceAddresses, RpcRequest rpcRequest) {
         int identityHashCode = System.identityHashCode(serviceAddresses);
+        // build rpc service name by rpcRequest
+        String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName();
 
         ConsistentHashSelector selector = selectors.get(rpcServiceName);
 
@@ -33,7 +37,7 @@ protected String doSelect(List<String> serviceAddresses, String rpcServiceName)
             selector = selectors.get(rpcServiceName);
         }
 
-        return selector.select(rpcServiceName);
+        return selector.select(rpcServiceName+ Arrays.stream(rpcRequest.getParameters()));
     }
 
     static class ConsistentHashSelector {
@@ -73,8 +77,8 @@ static long hash(byte[] digest, int idx) {
             return ((long) (digest[3 + idx * 4] & 255) << 24 | (long) (digest[2 + idx * 4] & 255) << 16 | (long) (digest[1 + idx * 4] & 255) << 8 | (long) (digest[idx * 4] & 255)) & 4294967295L;
         }
 
-        public String select(String rpcServiceName) {
-            byte[] digest = md5(rpcServiceName);
+        public String select(String rpcServiceKey) {
+            byte[] digest = md5(rpcServiceKey);
             return selectForKey(hash(digest, 0));
         }
 
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/loadbalancer/RandomLoadBalance.java b/rpc-framework-simple/src/main/java/github/javaguide/loadbalance/loadbalancer/RandomLoadBalance.java
@@ -1,6 +1,7 @@
 package github.javaguide.loadbalance.loadbalancer;
 
 import github.javaguide.loadbalance.AbstractLoadBalance;
+import github.javaguide.remoting.dto.RpcRequest;
 
 import java.util.List;
 import java.util.Random;
@@ -13,7 +14,7 @@
  */
 public class RandomLoadBalance extends AbstractLoadBalance {
     @Override
-    protected String doSelect(List<String> serviceAddresses, String rpcServiceName) {
+    protected String doSelect(List<String> serviceAddresses, RpcRequest rpcRequest) {
         Random random = new Random();
         return serviceAddresses.get(random.nextInt(serviceAddresses.size()));
     }
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/registry/ServiceDiscovery.java b/rpc-framework-simple/src/main/java/github/javaguide/registry/ServiceDiscovery.java
@@ -1,6 +1,7 @@
 package github.javaguide.registry;
 
 import github.javaguide.extension.SPI;
+import github.javaguide.remoting.dto.RpcRequest;
 
 import java.net.InetSocketAddress;
 
@@ -15,8 +16,8 @@ public interface ServiceDiscovery {
     /**
      * lookup service by rpcServiceName
      *
-     * @param rpcServiceName rpc service name
+     * @param rpcRequest rpc service pojo
      * @return service address
      */
-    InetSocketAddress lookupService(String rpcServiceName);
+    InetSocketAddress lookupService(RpcRequest rpcRequest);
 }
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/registry/zk/ZkServiceDiscovery.java b/rpc-framework-simple/src/main/java/github/javaguide/registry/zk/ZkServiceDiscovery.java
@@ -6,6 +6,7 @@
 import github.javaguide.loadbalance.LoadBalance;
 import github.javaguide.registry.ServiceDiscovery;
 import github.javaguide.registry.zk.util.CuratorUtils;
+import github.javaguide.remoting.dto.RpcRequest;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.curator.framework.CuratorFramework;
 
@@ -27,14 +28,15 @@ public ZkServiceDiscovery() {
     }
 
     @Override
-    public InetSocketAddress lookupService(String rpcServiceName) {
+    public InetSocketAddress lookupService(RpcRequest rpcRequest) {
+        String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName();
         CuratorFramework zkClient = CuratorUtils.getZkClient();
         List<String> serviceUrlList = CuratorUtils.getChildrenNodes(zkClient, rpcServiceName);
         if (serviceUrlList == null || serviceUrlList.size() == 0) {
             throw new RpcException(RpcErrorMessageEnum.SERVICE_CAN_NOT_BE_FOUND, rpcServiceName);
         }
         // load balancing
-        String targetServiceUrl = loadBalance.selectServiceAddress(serviceUrlList, rpcServiceName);
+        String targetServiceUrl = loadBalance.selectServiceAddress(serviceUrlList, rpcRequest);
         log.info("Successfully found the service address:[{}]", targetServiceUrl);
         String[] socketAddressArray = targetServiceUrl.split(":");
         String host = socketAddressArray[0];
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyRpcClient.java b/rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/netty/client/NettyRpcClient.java
@@ -98,9 +98,9 @@ public Object sendRpcRequest(RpcRequest rpcRequest) {
         // build return value
         CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
         // build rpc service name by rpcRequest
-        String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName();
+//        String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName();
         // get server address
-        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName);
+        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest);
         // get  server address related channel
         Channel channel = getChannel(inetSocketAddress);
         if (channel.isActive()) {
diff --git a/rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/socket/SocketRpcClient.java b/rpc-framework-simple/src/main/java/github/javaguide/remoting/transport/socket/SocketRpcClient.java
@@ -35,7 +35,7 @@ public Object sendRpcRequest(RpcRequest rpcRequest) {
         // build rpc service name by rpcRequest
         String rpcServiceName = RpcServiceProperties.builder().serviceName(rpcRequest.getInterfaceName())
                 .group(rpcRequest.getGroup()).version(rpcRequest.getVersion()).build().toRpcServiceName();
-        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName);
+        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest);
         try (Socket socket = new Socket()) {
             socket.connect(inetSocketAddress);
             ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
diff --git a/rpc-framework-simple/src/test/java/github/javaguide/loadbalance/loadbalancer/ConsistentHashLoadBalanceTest.java b/rpc-framework-simple/src/test/java/github/javaguide/loadbalance/loadbalancer/ConsistentHashLoadBalanceTest.java
@@ -2,11 +2,13 @@
 
 import github.javaguide.extension.ExtensionLoader;
 import github.javaguide.loadbalance.LoadBalance;
+import github.javaguide.remoting.dto.RpcRequest;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.UUID;
 
 import static org.junit.Assert.assertEquals;
 
@@ -17,10 +19,29 @@ void TestConsistentHashLoadBalance() {
         LoadBalance loadBalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension("loadBalance");
         List<String> serviceUrlList = new ArrayList<>(Arrays.asList("127.0.0.1:9997", "127.0.0.1:9998", "127.0.0.1:9999"));
         String userRpcServiceName = "github.javaguide.UserServicetest1version1";
-        String userServiceAddress = loadBalance.selectServiceAddress(serviceUrlList, userRpcServiceName);
+        //build rpcCall
+        RpcRequest rpcRequest = RpcRequest.builder()
+//                .parameters(args)
+                .interfaceName(userRpcServiceName)
+//                .paramTypes(method.getParameterTypes())
+                .requestId(UUID.randomUUID().toString())
+                .group("test2")
+                .version("version2")
+                .build();
+        String userServiceAddress = loadBalance.selectServiceAddress(serviceUrlList, rpcRequest);
         assertEquals("127.0.0.1:9999",userServiceAddress);
+
+
         String schoolRpcServiceName = "github.javaguide.SchoolServicetest1version1";
-        String schoolServiceAddress = loadBalance.selectServiceAddress(serviceUrlList, schoolRpcServiceName);
+        rpcRequest = RpcRequest.builder()
+//                .parameters(args)
+                .interfaceName(userRpcServiceName)
+//                .paramTypes(method.getParameterTypes())
+                .requestId(UUID.randomUUID().toString())
+                .group("test2")
+                .version("version2")
+                .build();
+        String schoolServiceAddress = loadBalance.selectServiceAddress(serviceUrlList, rpcRequest);
         assertEquals("127.0.0.1:9997",schoolServiceAddress);
     }
 }
diff --git a/rpc-framework-simple/src/test/java/github/javaguide/registry/ZkServiceRegistryTest.java b/rpc-framework-simple/src/test/java/github/javaguide/registry/ZkServiceRegistryTest.java
@@ -2,9 +2,11 @@
 
 import github.javaguide.registry.zk.ZkServiceDiscovery;
 import github.javaguide.registry.zk.ZkServiceRegistry;
+import github.javaguide.remoting.dto.RpcRequest;
 import org.junit.jupiter.api.Test;
 
 import java.net.InetSocketAddress;
+import java.util.UUID;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
@@ -21,7 +23,15 @@ void should_register_service_successful_and_lookup_service_by_service_name() {
         InetSocketAddress givenInetSocketAddress = new InetSocketAddress("127.0.0.1", 9333);
         zkServiceRegistry.registerService("github.javaguide.registry.zk.ZkServiceRegistry", givenInetSocketAddress);
         ServiceDiscovery zkServiceDiscovery = new ZkServiceDiscovery();
-        InetSocketAddress acquiredInetSocketAddress = zkServiceDiscovery.lookupService("github.javaguide.registry.zk.ZkServiceRegistry");
+        RpcRequest rpcRequest = RpcRequest.builder()
+//                .parameters(args)
+                .interfaceName("github.javaguide.registry.zk.ZkServiceRegistry")
+//                .paramTypes(method.getParameterTypes())
+                .requestId(UUID.randomUUID().toString())
+                .group("test2")
+                .version("version2")
+                .build();
+        InetSocketAddress acquiredInetSocketAddress = zkServiceDiscovery.lookupService(rpcRequest);
         assertEquals(givenInetSocketAddress.toString(), acquiredInetSocketAddress.toString());
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,7 @@`
`1`	`1`	`package github.javaguide.loadbalance;`
`2`	`2`
	`3`	`+import github.javaguide.remoting.dto.RpcRequest;`
	`4`	`+`
`3`	`5`	`import java.util.List;`
`4`	`6`
`5`	`7`	`/**`
`@@ -10,16 +12,16 @@`
`10`	`12`	`*/`
`11`	`13`	`public abstract class AbstractLoadBalance implements LoadBalance {`
`12`	`14`	`@Override`
`13`		`- public String selectServiceAddress(List<String> serviceAddresses, String rpcServiceName) {`
	`15`	`+ public String selectServiceAddress(List<String> serviceAddresses, RpcRequest rpcRequest) {`
`14`	`16`	`if (serviceAddresses == null \|\| serviceAddresses.size() == 0) {`
`15`	`17`	`return null;`
`16`	`18`	`}`
`17`	`19`	`if (serviceAddresses.size() == 1) {`
`18`	`20`	`return serviceAddresses.get(0);`
`19`	`21`	`}`
`20`		`- return doSelect(serviceAddresses, rpcServiceName);`
	`22`	`+ return doSelect(serviceAddresses, rpcRequest);`
`21`	`23`	`}`
`22`	`24`
`23`		`- protected abstract String doSelect(List<String> serviceAddresses, String rpcServiceName);`
	`25`	`+ protected abstract String doSelect(List<String> serviceAddresses, RpcRequest rpcRequest);`
`24`	`26`
`25`	`27`	`}`