前言

RPC 服务中,每个服务的容量都是有限的,即资源有限,只能承受住给定的网络请求,所以,在设计 RPC 框架的时候,一定要考虑流量控制这个问题。而 Java 中,实现流量控制有很多中方式,今天说 2 种。

Semaphore 实现流控

代码:

  static Semaphore semaphore = new Semaphore(100);

  public static void main(String[] args) {

    Executor timeTask = Executors.newScheduledThreadPool(1);
((ScheduledExecutorService) timeTask).scheduleAtFixedRate(
() -> semaphore.release(100 - semaphore.availablePermits()), 1000, 1000,
TimeUnit.MILLISECONDS); Executor pool = Executors.newFixedThreadPool(100); for (int i = 0; i < 100; i++) {
final int num = i;
pool.execute(() -> {
for (; ; ) {
for (int j = 0; j < 200; j++) {
if (semaphore.tryAcquire()) {
callRpc(num, j);
} else {
System.err.println("call fail");
}
}
}
});
}
} private static void callRpc(int num, int j) {
System.out.println(String.format("%s - %s: %d %d", new Date(), Thread.currentThread(), num, j));
}

代码中,我们模拟了 100 个线程,每个线程无限调用 RPC。

同时使用另一个定时任务,定时更新 Semaphore 可用许可为 100。

客户端线程调用时,会尝试获取信号量,当获取成功时,才会调用调用 RPC,反之,打印失败。

这个小程序实现了每秒钟限制 100 个请求的 RPC 的流量控制。

AtomicInteger 实现流控

代码:

  static AtomicInteger count = new AtomicInteger();

  public static void main(String[] args) {
Executor timeTask = Executors.newScheduledThreadPool(1);
((ScheduledExecutorService) timeTask).scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
count.getAndSet(100);
}
}, 1000, 1000, TimeUnit.MILLISECONDS); Executor pool = Executors.newFixedThreadPool(100); for (int i = 0; i < 100; i++) {
final int num = i;
pool.execute(() -> {
for (; ; ) {
for (int j = 0; j < 200; j++) {
if (count.get() >= 0) {// 快速判断,否则大量的 CAS 操作将会定时任务更新计数器 count
if (count.decrementAndGet() >= 0) {
callRpc(num, j);
}
}
}
}
});
}
} private static void callRpc(int num, int j) {
System.out.println(String.format("%s - %s: %d %d", new Date(), Thread.currentThread(), num, j));
}

这段代码和上面的类似,只是使用的 API 不同,这里使用的是 CAS。通过对 CAS 递减,达到流控的目的。

注意,这里有一个双重判断,先判断 count.get() >= 0,为什么呢?

如果直接使用 decrementAndGet 方法,则会使用 CAS,100 个线程并发使用 CAS ,将会导致定时任务的 CAS 操作不够及时。

所以,先判断,是否小于0 ,如果小于0了,就不必尝试 CAS,避免影响定时任务。

使用并发工具实现 RPC 调用流量控制的更多相关文章

  1. 《精通并发与Netty》学习笔记(07 - 基于Thrift实现Java与Python的RPC调用)

    上节我们介绍了基于Thrift实现java与java的RPC调用,本节我们基于Thrift实现Java与Python的RPC调用 首先,修改data.thirft文件,将命名空间由java改为py n ...

  2. java中的并发工具类

    在jdk的并发包里提供了几个非常有用的并发工具类.CountDownLatdch.CyclicBarrier和Semaphore工具类提供了一种并发流程控制的手段,Exchanger工具类则提供了在线 ...

  3. Java中的4个并发工具类 CountDownLatch CyclicBarrier Semaphore Exchanger

    在 java.util.concurrent 包中提供了 4 个有用的并发工具类 CountDownLatch 允许一个或多个线程等待其他线程完成操作,课题点 Thread 类的 join() 方法 ...

  4. Java中的并发工具类(CountDownLatch、CyclicBarrier、Semaphore、Exchanger)

    在JDK的并发包里提供了很多有意思的并发工具类.CountDownLatch.CyclicBarrier和Semaphore 工具类提供了一种并发流程控制的手段,Exchanger 工具类则提供了在线 ...

  5. 并发工具类(三)控制并发线程的数量 Semphore

    前言   JDK中为了处理线程之间的同步问题,除了提供锁机制之外,还提供了几个非常有用的并发工具类:CountDownLatch.CyclicBarrier.Semphore.Exchanger.Ph ...

  6. 基于AQS实现的Java并发工具类

    本文主要介绍一下基于AQS实现的Java并发工具类的作用,然后简单谈一下该工具类的实现原理.其实都是AQS的相关知识,只不过在AQS上包装了一下而已.本文也是基于您在有AQS的相关知识基础上,进行讲解 ...

  7. 25.大白话说java并发工具类-CountDownLatch,CyclicBarrier,Semaphore,Exchanger

    1. 倒计时器CountDownLatch 在多线程协作完成业务功能时,有时候需要等待其他多个线程完成任务之后,主线程才能继续往下执行业务功能,在这种的业务场景下,通常可以使用Thread类的join ...

  8. java并发之并发工具

    在JDK的并发包里提供了几个非常有用的并发工具类.CountDownLatch.CyclicBarrier和Semaphore工具类提供了一种并发流程控制的手段,Exchanger工具类则提供了在线程 ...

  9. Java 并发系列之八:java 并发工具(4个)

    1. CountDownLatch 2. CyclicBarrier 3. Semaphore 4. Exchanger 5. txt java 并发工具 通俗理解 CountDownLatch 等A ...

随机推荐

  1. 蓝桥杯---剪格子(DFS&amp;BFS)(小总结)

    问题描述 如下图所示,3 x 3 的格子中填写了一些整数. +--*--+--+ |10* 1|52| +--****--+ |20|30* 1| *******--+ | 1| 2| 3| +--+ ...

  2. 有没有好用的开源sql语法分析器? - 匿名用户的回答 - 知乎

    有没有好用的开源sql语法分析器? - 匿名用户的回答 - 知乎 presto,hive,drill,calcite,sparksq

  3. res里面的drawable(ldpi、mdpi、hdpi、xhdpi、xxhdpi)

    (1)drawable-hdpi里面存放高分辨率的图片,如WVGA (480x800),FWVGA (480x854) (2)drawable-mdpi里面存放中等分辨率的图片,如HVGA (320x ...

  4. bash启动加载过程

  5. logback中批量插入数据库的参考代码

    protected void insertProperties(Map<String, String> mergedMap, Connection connection, long eve ...

  6. python笔记18-高阶函数

    高阶函数: 如果一个函数的入参是一个函数名的话,那这个函数就是一个高阶函数 函数即变量 # def hello(name):# print(name)# new_hello = hello#hello ...

  7. json_decode与json_encode容易被忽视的点

    一.json_decode的使用,json_decode共有4个参数 json_decode ( string $json [, bool $assoc=FALSE [, int $depth= 51 ...

  8. 定时调度篇之Quartz.Net详解(被替换)

    一. 背景 我们在日常开发中,可能你会遇到这样的需求:"每个月的3号给用户发信息,提醒用户XXX "."每天的0点需要统计前一天的考勤记录"."每个月 ...

  9. 使用idea创建一个maven工程

    使用idea创建一个maven工程 一.创建maven工程 二.输入工程名 三.指定maven仓库 四.点击finish 五.修改工程结构(file->project structure) 六. ...

  10. 使用synchronized wait() notifyall() 实现简单的加减法同步 竞争抢答

    import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.co ...