add zk

Author: kimmking

07rpc/rpc01/rpcfx-core/pom.xml

@@ -24,12 +24,31 @@
  <version>1.2.70</version>
  </dependency>
 
+ <dependency>
+ <groupId>org.projectlombok</groupId>
+ <artifactId>lombok</artifactId>
+ <version>1.18.16</version>
+ </dependency>
+
  <dependency>
  <groupId>com.squareup.okhttp3</groupId>
  <artifactId>okhttp</artifactId>
  <version>3.12.2</version>
  </dependency>
 
+
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-client</artifactId>
+ <version>5.1.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-framework</artifactId>
+ <version>5.1.0</version>
+ </dependency>
+
  <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter</artifactId>

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/api/Filter.java

@@ -4,6 +4,6 @@ public interface Filter{
 
 booleanfilter(RpcfxRequestrequest);
 
-Filternext();
+// Filter next();
 
 }

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/api/RpcfxRequest.java

@@ -1,34 +1,10 @@
 packageio.kimmking.rpcfx.api;
 
-publicclassRpcfxRequest{
+importlombok.Data;
 
+@Data
+publicclassRpcfxRequest{
 privateStringserviceClass;
-
 privateStringmethod;
-
 privateObject[] params;
-
-publicStringgetServiceClass(){
-returnserviceClass;
- }
-
-publicvoidsetServiceClass(StringserviceClass){
-this.serviceClass = serviceClass;
- }
-
-publicStringgetMethod(){
-returnmethod;
- }
-
-publicvoidsetMethod(Stringmethod){
-this.method = method;
- }
-
-publicObject[] getParams(){
-returnparams;
- }
-
-publicvoidsetParams(Object[] params){
-this.params = params;
- }
 }

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/api/RpcfxResponse.java

@@ -1,34 +1,10 @@
 packageio.kimmking.rpcfx.api;
 
-publicclassRpcfxResponse{
+importlombok.Data;
 
+@Data
+publicclassRpcfxResponse{
 privateObjectresult;
-
 privatebooleanstatus;
-
 privateExceptionexception;
-
-publicObjectgetResult(){
-returnresult;
- }
-
-publicvoidsetResult(Objectresult){
-this.result = result;
- }
-
-publicbooleanisStatus(){
-returnstatus;
- }
-
-publicvoidsetStatus(booleanstatus){
-this.status = status;
- }
-
-publicExceptiongetException(){
-returnexception;
- }
-
-publicvoidsetException(Exceptionexception){
-this.exception = exception;
- }
 }

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/api/ServiceProviderDesc.java

@@ -0,0 +1,16 @@
+packageio.kimmking.rpcfx.api;
+
+importlombok.Builder;
+importlombok.Data;
+
+@Data
+@Builder
+publicclassServiceProviderDesc{
+
+privateStringhost;
+privateIntegerport;
+privateStringserviceClass;
+
+// group
+// version
+}

07rpc/rpc01/rpcfx-core/src/main/java/io/kimmking/rpcfx/client/Rpcfx.java

@@ -37,10 +37,10 @@ public static <T, filters> T createFromRegistry(final Class<T> serviceClass, fin
 
  }
 
-publicstatic <T> Tcreate(finalClass<T> serviceClass, finalStringurl, Filterfilter){
+publicstatic <T> Tcreate(finalClass<T> serviceClass, finalStringurl, Filter... filters){
 
 // 0. 替换动态代理 -> AOP
-return (T) Proxy.newProxyInstance(Rpcfx.class.getClassLoader(), newClass[]{serviceClass}, newRpcfxInvocationHandler(serviceClass, url, filter));
+return (T) Proxy.newProxyInstance(Rpcfx.class.getClassLoader(), newClass[]{serviceClass}, newRpcfxInvocationHandler(serviceClass, url, filters));
 
  }
 
@@ -73,9 +73,11 @@ public Object invoke(Object proxy, Method method, Object[] params) throws Throwa
 request.setMethod(method.getName());
 request.setParams(params);
 
-for (Filterfilter : filters){
-if(!filter.filter(request)){
-returnnull;
+if (null!=filters){
+for (Filterfilter : filters){
+if (!filter.filter(request)){
+returnnull;
+ }
  }
  }
 

07rpc/rpc01/rpcfx-demo-consumer/src/main/java/io/kimmking/rpcfx/demo/consumer/RpcfxClientApplication.java

@@ -1,12 +1,18 @@
 packageio.kimmking.rpcfx.demo.consumer;
 
+importio.kimmking.rpcfx.api.Filter;
+importio.kimmking.rpcfx.api.LoadBalancer;
+importio.kimmking.rpcfx.api.Router;
+importio.kimmking.rpcfx.api.RpcfxRequest;
 importio.kimmking.rpcfx.client.Rpcfx;
 importio.kimmking.rpcfx.demo.api.Order;
 importio.kimmking.rpcfx.demo.api.OrderService;
 importio.kimmking.rpcfx.demo.api.User;
 importio.kimmking.rpcfx.demo.api.UserService;
+importlombok.extern.slf4j.Slf4j;
 importorg.springframework.boot.autoconfigure.SpringBootApplication;
 
+importjava.util.List;
 importjava.util.Random;
 
 @SpringBootApplication
@@ -36,6 +42,28 @@ public static void main(String[] args){
 // SpringApplication.run(RpcfxClientApplication.class, args);
  }
 
+privatestaticclassTagRouterimplementsRouter{
+@Override
+publicList<String> route(List<String> urls){
+returnurls;
+ }
+ }
+
+privatestaticclassRandomLoadBalancerimplementsLoadBalancer{
+@Override
+publicStringselect(List<String> urls){
+returnurls.get(0);
+ }
+ }
+
+@Slf4j
+privatestaticclassCuicuiFilterimplementsFilter{
+@Override
+publicbooleanfilter(RpcfxRequestrequest){
+log.info("filter{} ->{}", this.getClass().getName(), request.toString());
+returntrue;
+ }
+ }
 }
 
 

07rpc/rpc01/rpcfx-demo-provider/pom.xml

@@ -38,12 +38,6 @@
  <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
 
- <dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-client</artifactId>
- <version>5.1.0</version>
- </dependency>
-
  <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-test</artifactId>

07rpc/rpc01/rpcfx-demo-provider/src/main/java/io/kimmking/rpcfx/demo/provider/RpcfxServerApplication.java

@@ -1,11 +1,18 @@
 packageio.kimmking.rpcfx.demo.provider;
 
+importcom.alibaba.fastjson.JSON;
 importio.kimmking.rpcfx.api.RpcfxRequest;
 importio.kimmking.rpcfx.api.RpcfxResolver;
 importio.kimmking.rpcfx.api.RpcfxResponse;
+importio.kimmking.rpcfx.api.ServiceProviderDesc;
 importio.kimmking.rpcfx.demo.api.OrderService;
 importio.kimmking.rpcfx.demo.api.UserService;
 importio.kimmking.rpcfx.server.RpcfxInvoker;
+importorg.apache.curator.RetryPolicy;
+importorg.apache.curator.framework.CuratorFramework;
+importorg.apache.curator.framework.CuratorFrameworkFactory;
+importorg.apache.curator.retry.ExponentialBackoffRetry;
+importorg.apache.zookeeper.CreateMode;
 importorg.springframework.beans.factory.annotation.Autowired;
 importorg.springframework.boot.SpringApplication;
 importorg.springframework.boot.autoconfigure.SpringBootApplication;
@@ -24,18 +31,44 @@ public class RpcfxServerApplication{
 
 publicstaticvoidmain(String[] args) throwsException{
 
+// start zk client
+RetryPolicyretryPolicy = newExponentialBackoffRetry(1000, 3);
+CuratorFrameworkclient = CuratorFrameworkFactory.builder().connectString("localhost:2181").namespace("rpcfx").retryPolicy(retryPolicy).build();
+client.start();
+
+
+// register service
 // xxx "io.kimmking.rpcfx.demo.api.UserService"
 
-ServiceProviderDescdesc = newServiceProviderDesc();
-desc.setHost(InetAddress.getLocalHost().getHostAddress());
-desc.setPort(8080);
-desc.setServiceClass("io.kimmking.rpcfx.demo.api.UserService");
+StringuserService = "io.kimking.rpcfx.demo.api.UserService";
+registerService(client, userService);
+StringorderService = "io.kimking.rpcfx.demo.api.OrderService";
+registerService(client, orderService);
+
 
-// Curator.
+// 进一步的优化，是在spring加载完成后，从里面拿到特定注解的bean，自动注册到zk
 
 SpringApplication.run(RpcfxServerApplication.class, args);
  }
 
+privatestaticvoidregisterService(CuratorFrameworkclient, Stringservice) throwsException{
+ServiceProviderDescuserServiceSesc = ServiceProviderDesc.builder()
+ .host(InetAddress.getLocalHost().getHostAddress())
+ .port(8080).serviceClass(service).build();
+// String userServiceSescJson = JSON.toJSONString(userServiceSesc);
+
+try{
+if ( null == client.checkExists().forPath("/" + service)){
+client.create().withMode(CreateMode.PERSISTENT).forPath("/" + service, "service".getBytes());
+ }
+ } catch (Exceptionex){
+ex.printStackTrace();
+ }
+
+client.create().withMode(CreateMode.EPHEMERAL).
+forPath( "/" + service + "/" + userServiceSesc.getHost() + "_" + userServiceSesc.getPort(), "provider".getBytes());
+ }
+
 @Autowired
 RpcfxInvokerinvoker;