Skip to content

Commit 097bedd

Browse files
LbxxxLbxxx
authored
Merge pull request #4 from JavaCourse00/main
1
2 parents fc4e769 + bb818e2 commit 097bedd

39 files changed

Lines changed: 4828 additions & 64 deletions

File tree

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,31 @@
2424
<version>1.2.70</version>
2525
</dependency>
2626

27+
<dependency>
28+
<groupId>org.projectlombok</groupId>
29+
<artifactId>lombok</artifactId>
30+
<version>1.18.16</version>
31+
</dependency>
32+
2733
<dependency>
2834
<groupId>com.squareup.okhttp3</groupId>
2935
<artifactId>okhttp</artifactId>
3036
<version>3.12.2</version>
3137
</dependency>
3238

39+
40+
<dependency>
41+
<groupId>org.apache.curator</groupId>
42+
<artifactId>curator-client</artifactId>
43+
<version>5.1.0</version>
44+
</dependency>
45+
46+
<dependency>
47+
<groupId>org.apache.curator</groupId>
48+
<artifactId>curator-framework</artifactId>
49+
<version>5.1.0</version>
50+
</dependency>
51+
3352
<dependency>
3453
<groupId>org.springframework.boot</groupId>
3554
<artifactId>spring-boot-starter</artifactId>

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package io.kimmking.rpcfx.api;
2+
3+
public interface Filter {
4+
5+
boolean filter(RpcfxRequest request);
6+
7+
// Filter next();
8+
9+
}

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package io.kimmking.rpcfx.api;
2+
3+
import java.util.List;
4+
5+
public interface LoadBalancer {
6+
7+
String select(List<String> urls);
8+
9+
}

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package io.kimmking.rpcfx.api;
2+
3+
import java.util.List;
4+
5+
public interface Router {
6+
7+
List<String> route(List<String> urls);
8+
}

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

Lines changed: 3 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,10 @@
11
package io.kimmking.rpcfx.api;
22

3-
public class RpcfxRequest {
3+
import lombok.Data;
44

5+
@Data
6+
public class RpcfxRequest {
57
private String serviceClass;
6-
78
private String method;
8-
99
private Object[] params;
10-
11-
public String getServiceClass() {
12-
return serviceClass;
13-
}
14-
15-
public void setServiceClass(String serviceClass) {
16-
this.serviceClass = serviceClass;
17-
}
18-
19-
public String getMethod() {
20-
return method;
21-
}
22-
23-
public void setMethod(String method) {
24-
this.method = method;
25-
}
26-
27-
public Object[] getParams() {
28-
return params;
29-
}
30-
31-
public void setParams(Object[] params) {
32-
this.params = params;
33-
}
3410
}

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

Lines changed: 3 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,10 @@
11
package io.kimmking.rpcfx.api;
22

3-
public class RpcfxResponse {
3+
import lombok.Data;
44

5+
@Data
6+
public class RpcfxResponse {
57
private Object result;
6-
78
private boolean status;
8-
99
private Exception exception;
10-
11-
public Object getResult() {
12-
return result;
13-
}
14-
15-
public void setResult(Object result) {
16-
this.result = result;
17-
}
18-
19-
public boolean isStatus() {
20-
return status;
21-
}
22-
23-
public void setStatus(boolean status) {
24-
this.status = status;
25-
}
26-
27-
public Exception getException() {
28-
return exception;
29-
}
30-
31-
public void setException(Exception exception) {
32-
this.exception = exception;
33-
}
3410
}

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package io.kimmking.rpcfx.api;
2+
3+
import lombok.Builder;
4+
import lombok.Data;
5+
6+
@Data
7+
@Builder
8+
public class ServiceProviderDesc {
9+
10+
private String host;
11+
private Integer port;
12+
private String serviceClass;
13+
14+
// group
15+
// version
16+
}

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

Lines changed: 41 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,7 @@
33

44
import com.alibaba.fastjson.JSON;
55
import com.alibaba.fastjson.parser.ParserConfig;
6-
import io.kimmking.rpcfx.api.RpcfxRequest;
7-
import io.kimmking.rpcfx.api.RpcfxResponse;
6+
import io.kimmking.rpcfx.api.*;
87
import okhttp3.MediaType;
98
import okhttp3.OkHttpClient;
109
import okhttp3.Request;
@@ -14,17 +13,36 @@
1413
import java.lang.reflect.InvocationHandler;
1514
import java.lang.reflect.Method;
1615
import java.lang.reflect.Proxy;
16+
import java.util.ArrayList;
17+
import java.util.List;
1718

1819
public final class Rpcfx {
1920

2021
static {
2122
ParserConfig.getGlobalInstance().addAccept("io.kimmking");
2223
}
2324

24-
public static <T> T create(final Class<T> serviceClass, final String url) {
25+
public static <T, filters> T createFromRegistry(final Class<T> serviceClass, final String zkUrl, Router router, LoadBalancer loadBalance, Filter filter) {
26+
27+
// 加filte之一
28+
29+
// curator Provider list from zk
30+
List<String> invokers = new ArrayList<>();
31+
// 1. 简单：从zk拿到服务提供的列表
32+
// 2. 挑战：监听zk的临时节点，根据事件更新这个list（注意，需要做个全局map保持每个服务的提供者List）
33+
34+
List<String> urls = router.route(invokers);
35+
36+
String url = loadBalance.select(urls); // router, loadbalance
37+
38+
return (T) create(serviceClass, url, filter);
39+
40+
}
41+
42+
public static <T> T create(final Class<T> serviceClass, final String url, Filter... filters) {
2543

2644
// 0. 替换动态代理 -> AOP
27-
return (T) Proxy.newProxyInstance(Rpcfx.class.getClassLoader(), new Class[]{serviceClass}, new RpcfxInvocationHandler(serviceClass, url));
45+
return (T) Proxy.newProxyInstance(Rpcfx.class.getClassLoader(), new Class[]{serviceClass}, new RpcfxInvocationHandler(serviceClass, url, filters));
2846

2947
}
3048

@@ -34,9 +52,12 @@ public static class RpcfxInvocationHandler implements InvocationHandler {
3452

3553
private final Class<?> serviceClass;
3654
private final String url;
37-
public <T> RpcfxInvocationHandler(Class<T> serviceClass, String url) {
55+
private final Filter[] filters;
56+
57+
public <T> RpcfxInvocationHandler(Class<T> serviceClass, String url, Filter... filters) {
3858
this.serviceClass = serviceClass;
3959
this.url = url;
60+
this.filters = filters;
4061
}
4162

4263
// 可以尝试，自己去写对象序列化，二进制还是文本的，，，rpcfx是xml自定义序列化、反序列化，json: code.google.com/p/rpcfx
@@ -45,13 +66,28 @@ public <T> RpcfxInvocationHandler(Class<T> serviceClass, String url) {
4566

4667
@Override
4768
public Object invoke(Object proxy, Method method, Object[] params) throws Throwable {
69+
70+
// 加filter地方之二
71+
// mock == true, new Student("hubao");
72+
4873
RpcfxRequest request = new RpcfxRequest();
4974
request.setServiceClass(this.serviceClass.getName());
5075
request.setMethod(method.getName());
5176
request.setParams(params);
5277

78+
if (null!=filters) {
79+
for (Filter filter : filters) {
80+
if (!filter.filter(request)) {
81+
return null;
82+
}
83+
}
84+
}
85+
5386
RpcfxResponse response = post(request, url);
5487

88+
// 加filter地方之三
89+
// Student.setTeacher("cuijing");
90+
5591
// 这里判断response.status，处理异常
5692
// 考虑封装一个全局的RpcfxException
5793

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

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,20 @@
11
package io.kimmking.rpcfx.demo.consumer;
22

3+
import io.kimmking.rpcfx.api.Filter;
4+
import io.kimmking.rpcfx.api.LoadBalancer;
5+
import io.kimmking.rpcfx.api.Router;
6+
import io.kimmking.rpcfx.api.RpcfxRequest;
37
import io.kimmking.rpcfx.client.Rpcfx;
48
import io.kimmking.rpcfx.demo.api.Order;
59
import io.kimmking.rpcfx.demo.api.OrderService;
610
import io.kimmking.rpcfx.demo.api.User;
711
import io.kimmking.rpcfx.demo.api.UserService;
12+
import lombok.extern.slf4j.Slf4j;
813
import org.springframework.boot.autoconfigure.SpringBootApplication;
914

15+
import java.util.List;
16+
import java.util.Random;
17+
1018
@SpringBootApplication
1119
public class RpcfxClientApplication {
1220

@@ -28,9 +36,35 @@ public static void main(String[] args) {
2836
Order order = orderService.findOrderById(1992129);
2937
System.out.println(String.format("find order name=%s, amount=%f",order.getName(),order.getAmount()));
3038

31-
// 新加一个OrderService
39+
//
40+
UserService userService2 = Rpcfx.createFromRegistry(UserService.class, "localhost:2181", new TagRouter(), new RandomLoadBalancer(), new CuicuiFilter());
3241

3342
// SpringApplication.run(RpcfxClientApplication.class, args);
3443
}
3544

45+
private static class TagRouter implements Router {
46+
@Override
47+
public List<String> route(List<String> urls) {
48+
return urls;
49+
}
50+
}
51+
52+
private static class RandomLoadBalancer implements LoadBalancer {
53+
@Override
54+
public String select(List<String> urls) {
55+
return urls.get(0);
56+
}
57+
}
58+
59+
@Slf4j
60+
private static class CuicuiFilter implements Filter {
61+
@Override
62+
public boolean filter(RpcfxRequest request) {
63+
log.info("filter {} -> {}", this.getClass().getName(), request.toString());
64+
return true;
65+
}
66+
}
3667
}
68+
69+
70+

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

Lines changed: 52 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,18 @@
11
package io.kimmking.rpcfx.demo.provider;
22

3+
import com.alibaba.fastjson.JSON;
34
import io.kimmking.rpcfx.api.RpcfxRequest;
45
import io.kimmking.rpcfx.api.RpcfxResolver;
56
import io.kimmking.rpcfx.api.RpcfxResponse;
7+
import io.kimmking.rpcfx.api.ServiceProviderDesc;
68
import io.kimmking.rpcfx.demo.api.OrderService;
79
import io.kimmking.rpcfx.demo.api.UserService;
810
import io.kimmking.rpcfx.server.RpcfxInvoker;
11+
import org.apache.curator.RetryPolicy;
12+
import org.apache.curator.framework.CuratorFramework;
13+
import org.apache.curator.framework.CuratorFrameworkFactory;
14+
import org.apache.curator.retry.ExponentialBackoffRetry;
15+
import org.apache.zookeeper.CreateMode;
916
import org.springframework.beans.factory.annotation.Autowired;
1017
import org.springframework.boot.SpringApplication;
1118
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -14,14 +21,54 @@
1421
import org.springframework.web.bind.annotation.RequestBody;
1522
import org.springframework.web.bind.annotation.RestController;
1623

24+
import java.net.InetAddress;
25+
import java.net.InterfaceAddress;
26+
import java.net.UnknownHostException;
27+
1728
@SpringBootApplication
1829
@RestController
1930
public class RpcfxServerApplication {
2031

21-
public static void main(String[] args) {
32+
public static void main(String[] args) throws Exception {
33+
34+
// start zk client
35+
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
36+
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181").namespace("rpcfx").retryPolicy(retryPolicy).build();
37+
client.start();
38+
39+
40+
// register service
41+
// xxx "io.kimmking.rpcfx.demo.api.UserService"
42+
43+
String userService = "io.kimking.rpcfx.demo.api.UserService";
44+
registerService(client, userService);
45+
String orderService = "io.kimking.rpcfx.demo.api.OrderService";
46+
registerService(client, orderService);
47+
48+
49+
// 进一步的优化，是在spring加载完成后，从里面拿到特定注解的bean，自动注册到zk
50+
2251
SpringApplication.run(RpcfxServerApplication.class, args);
2352
}
2453

54+
private static void registerService(CuratorFramework client, String service) throws Exception {
55+
ServiceProviderDesc userServiceSesc = ServiceProviderDesc.builder()
56+
.host(InetAddress.getLocalHost().getHostAddress())
57+
.port(8080).serviceClass(service).build();
58+
// String userServiceSescJson = JSON.toJSONString(userServiceSesc);
59+
60+
try {
61+
if ( null == client.checkExists().forPath("/" + service)) {
62+
client.create().withMode(CreateMode.PERSISTENT).forPath("/" + service, "service".getBytes());
63+
}
64+
} catch (Exception ex) {
65+
ex.printStackTrace();
66+
}
67+
68+
client.create().withMode(CreateMode.EPHEMERAL).
69+
forPath( "/" + service + "/" + userServiceSesc.getHost() + "_" + userServiceSesc.getPort(), "provider".getBytes());
70+
}
71+
2572
@Autowired
2673
RpcfxInvoker invoker;
2774

@@ -42,6 +89,10 @@ public RpcfxResolver createResolver(){
4289

4390
// 能否去掉name
4491
//
92+
93+
// annotation
94+
95+
4596
@Bean(name = "io.kimmking.rpcfx.demo.api.UserService")
4697
public UserService createUserService(){
4798
return new UserServiceImpl();

0 commit comments

Comments
 (0)