【RPC】一步一步实现基于netty+zookeeper的RPC框架（五）

上一篇实现了服务的链路追踪,本篇带来限流。

关于服务限流，比较流行的是
1：令牌桶算法：桶算法的升级版，实现简单，应对热点请求效果更理想。
2：动态限流：根据实时的统计当前时间段请求响应时间来动态调整限流数量，实现复杂，但应对各种情况效果更好。

这里还是贴出github代码地址,想直接看代码的可以直接下载运行:https://github.com/whiteBX/wrpc

本篇带来令牌桶实现限流的实现，核心代码很简单，主要就下面一个类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
1public class ConsumerLimiter {

2

3    /**

4     * 存放限流器

5     */

6    private ConcurrentMap<String, RateLimiter> rateLimiterMap = new ConcurrentHashMap<String, RateLimiter>();

7

8    /**

9     * 限流

10     * @param appCode

11     * @return

12     */

13    public boolean limit(String appCode) {

14        RateLimiter rateLimiter = rateLimiterMap.get(appCode);

15        if (rateLimiter == null) {

16            rateLimiterMap.putIfAbsent(appCode, RateLimiter.create(1));

17        }

18        rateLimiter = rateLimiterMap.get(appCode);

19        // 默认等待500mS

20        return rateLimiter.tryAcquire(500, TimeUnit.NANOSECONDS);

21    }

22}

23

24

这里核心就是用到了guava包的RateLimiter，它内部实现了令牌桶，默认情况下RateLimiter.create(1)表示每秒向令牌桶中投放1个令牌，当然它也支持自己传入时间单位。

家下来改造RPCConsumer类的getServer方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
1/**

2 * 获取服务

3 * @param appCode

4 * @return

5 */

6private String getServer(String appCode) {

7    // 限流

8    if (!consumerLimiter.limit(appCode)) {

9        System.out.println("请求被限流");

10        return null;

11    }

12    // 从zookeeper获取服务地址

13    String serverHost = urlHolder.getUrl(appCode);

14    if (serverHost == null) {

15        return null;

16    }

17    return serverHost;

18}

19

20

运行我们的代码，客户端日志如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1zookeeper连接成功

2链路追踪，调用远程服务：{"appCode":"100000","localIp":"10.0.75.1","operationName":"hello","parentSpanId":"0","remoteIp":"127.0.0.1,8092","spanId":"5bd706ef9f703a049cf62dd7","timestamp":1540818671137,"traceId":"5bd706ef9f703a049cf62dd6"}

3调用服务器:127.0.0.1,8092,请求参数:{"clazzName":"org.white.wrpc.hello.service.HelloService","data":"{\"seq\":0}","methodName":"hello","paramTypeName":"org.white.wrpc.hello.model.request.HelloRequest","span":{"appCode":"100000","localIp":"10.0.75.1","operationName":"hello","parentSpanId":"0","remoteIp":"127.0.0.1,8092","spanId":"5bd706ef9f703a049cf62dd7","timestamp":1540818671137,"traceId":"5bd706ef9f703a049cf62dd6"}},响应参数:{"code":200,"message":"success:0"}

4客户端收到响应:{"code":200,"message":"success:0"}

5请求被限流

6链路追踪，调用远程服务：{"appCode":"100000","localIp":"10.0.75.1","operationName":"hello","parentSpanId":"0","spanId":"5bd706f19f703a049cf62ddd","timestamp":1540818673006,"traceId":"5bd706f19f703a049cf62ddc"}

7远程调用错误:当前无服务提供者

8客户端收到响应:{"code":404,"message":"no provider"}

9请求被限流

10链路追踪，调用远程服务：{"appCode":"100000","localIp":"10.0.75.1","operationName":"hello","parentSpanId":"0","spanId":"5bd706f19f703a049cf62ddf","timestamp":1540818673307,"traceId":"5bd706f19f703a049cf62dde"}

11远程调用错误:当前无服务提供者

12客户端收到响应:{"code":404,"message":"no provider"}

13链路追踪，调用远程服务：{"appCode":"100000","localIp":"10.0.75.1","operationName":"hello","parentSpanId":"0","remoteIp":"127.0.0.1,8093","spanId":"5bd706f19f703a049cf62de1","timestamp":1540818673609,"traceId":"5bd706f19f703a049cf62de0"}

14调用服务器:127.0.0.1,8093,请求参数:{"clazzName":"org.white.wrpc.hello.service.HelloService","data":"{\"seq\":5}","methodName":"hello","paramTypeName":"org.white.wrpc.hello.model.request.HelloRequest","span":{"appCode":"100000","localIp":"10.0.75.1","operationName":"hello","parentSpanId":"0","remoteIp":"127.0.0.1,8093","spanId":"5bd706f19f703a049cf62de1","timestamp":1540818673609,"traceId":"5bd706f19f703a049cf62de0"}},响应参数:{"code":200,"message":"success:5"}

15客户端收到响应:{"code":200,"message":"success:5"}

16请求被限流

17

18

可以看到其中的限流日志。

另外说一点，本篇实现的是客户端限流，也有很多是支持服务端限流的，各有各的好处，可以按各位实际场景决定，或者两者都用。