java 应用的并发与流量控制

前段时间在团队内部做了一次关于并发以及流量控制原理与实施的分享，今天花时间整理了下，分享给大家，同时也希望大家多提宝贵意见，共同进步。

说到java并发，不得不提一下java的线程模型

Java的线程模型

Java的并发实际上是在thread的基础上实现的，因此，说到并发限流的原理，不得不谈到java线程的五种状态之间的转换

图1 java线程的五种状态

从new状态到start，然后运行，running，当收到wait消息，或者sleep时，以及处理IO的时候，线程会进入Blocked状态，直到被notify，或者IO完成，当线程执行结束，会进入dead状态。

Java线程的实现实际上是跟操作系统的实现相关联的，不同的操作系统，有不同的线程模型，比如一对一模型，一对多模型，多对多模型，来完成用户态线程与内核态线程的对应。

一对一模型

多对一模型

多对多模型

当然，只有在running状态的线程，是占用cpu时间片的，而处于block状态的线程，会通过上下文切换，将线程状态信息存入内存的TCB(线程控制块)，这个过程中，会有一些寄存器的指令，程序计数器等等涉及线程运行状态的信息，保存到TCB中，操作系统会调度下一个运行状态的thread，将TCB状态还原，开始运行，等到时间片用完，又将线程切换出来。

上下文切换是纯粹调度的开销，可以讲是纯粹的时间浪费，但是这种浪费在所难免。

Samphore

先截取几段samphore实现的源代码，以代码说话更有说服力

 

 

图2 samphore的acquire方法调用栈

很容易可以发现，samphore的实现是忙等待的，也就是说，在等待资源释放的过程中，samphore的acquire操作，不断的在做死循环，而这个操作，是需要占用和消耗cpu时间片的，这种机制也称作为spinlock(自旋锁)。

大家可能奇怪，为什么要不断做死循环呢，难道这样不是浪费资源么？

其实，任何事情都存在两面性，线程状态的切换过程中，可能会带来上下文切换的开销，如果在循环一段时间后，samphore便有线程将资源释放，samphore会很快解锁，相对于上下文切换的开销来说，这个开销会小的多，也就是说，cpu时间片的浪费是值得的。但是，如果samphore所锁定的一段临界代码段，中间包含长时间的计算任务，亦或是磁盘，网络，DB等操作，忙等的时间可能相当长，这样，spinlock的死循环就相当不划算了，这种可以预估到的长时间的等待，还不如使用上下文切换，将线程block住，让其真正的等待来的划算。

乐观锁与悲观锁

接下来介绍下乐观锁与悲观锁，悲观锁认为，如果不采用加锁的同步操作，那么肯定会出问题，无论共享数据是否真的会出现竞争，先加锁再说。因此，每次都必须先加锁，再操作，操作完后解锁。典型的就是java的synchronized关键字，以及concurrent包下的一些lock。

悲观锁的实现其实可以包含以上两种思路，spinlock以及上下文切换方式，让线程等待。

而乐观锁恰恰相反，乐观锁是基于冲突检测的，通俗的说，就是先进行操作，如果没有其他线程争用数据，操作就成功了，如果有争用，最常见的补偿措施就是不断的进行重试，直到操作成功。典型的应用就是使用java原子类型的cas操作，不断的比较和设值。

原子操作的误区

大家觉得i++ 是原子操作么？

相信大部分人都知道，i++不能认为是原子操作，理由便是，i++操作可以分割成多条汇报指令，最终提交给cpu执行，如果恰好执行不在一个cpu时间片周期里，在执行的过程中，如果没有采取必要的同步措施，很可能会与其他线程的指令交叠执行，即便是单个线程，也可能遇到cpu的指令重排序。

图3 I++操作翻译成汇编指令(粗略的)

依赖于硬件的指令，Java对于原子操作提供了原生的支持，最典型的例子便是Atomic变量。

AtomicInteger. compareAndSet(int expect, int update)，实际上是依赖cpu提供的cmpxchgl s,d来支持的，该指令执行过程为，先将expact值送到eax寄存器当中，调用cmpxchal指令，update和*atomic作为其源操作数和目标操作数，最后将汇编指令执行的结果(eax寄存器中)返回，*atomic原子变量仅仅有可能被更新，当且仅当expect与真实值相等。

这样，通过cpu的原生支持，通过一条指令，即保证了操作的原子性。

流控

为什么要流控，其实很简单，任何应用都有一个设计指标，当应用的压力超过了他设计所能承载的能力时，就好比一座只允许行人通过的独木桥，是无法承载一辆坦克的重量的，这个时候，为了让机器能够继续运行，在不宕机的情况下尽其所能的对一部分用户提供服务，保证整个流程能够继续走下去，这个时候，就必须对应用进行流控，丢弃一部分用户的请无法避免。

流控可以从多个维度来进行，比如针对QPS，并发线程数，黑白名单，加权分级等等，最典型最直接的便是QPS和并发线程数的流控。当然，要进行流控，首先等有一个流控的阀值，这个阀值不是说拍拍脑袋就能够想出来，不同类型的应用，所面临的情况不一样，也没有一个统一的衡量标准，必须经过多轮的压力测试，才能够得出一个比较靠谱的数值。

图4 并发数限流和QPS限流对于系统load的影响

Qps限流的话在前半秒可能会有一个load上升，后半秒下降的这么一个波动过程，而对于并发数限流来说，整个系统load曲线会更加平稳。

 

 

使用semphore进行并发流控

 

    Semaphore semphore = new Semaphore(10);
    if(semphore.getQueueLength() > 10){
    	//等待队列阀值为10时
    	return;
    }
    try {
			semphore.acquire();
			
			//干活
			
		} catch (InterruptedException e) {
			e.printStackTrace();
		}finally{
			semphore.release();//释放
		}

 

 

 

使用乐观锁加上下文切换进行流控

 

	public void enter(Object obj){

		boolean isUpdate = false;
		int countValue = count.get();
		if(countValue > 0){
			isUpdate = count.compareAndSet(countValue, countValue -1); 
			if(isUpdate)return;
		}

		concurQueue.add(obj);

		try {
			obj.wait();
		} catch (InterruptedException e) {
			logger.error("flowcontrol thread was interrupted .......",e);
		}
		return ;
	}

	public void release(){

		synchronized(count){
			if(count.get() < VALVE){
				count.set(count.get() + 1);
			}
		}

		Object obj = concurQueue.remove();
		if(obj != null){
			synchronized (obj) {
				obj.notify();
			}
		}
		System.out.println("notify ...............");
		return ;
	}

 

 

具体采用信号量还是使用上下文切换形式，需要根据临界代码段执行的时间而定

 

上面说了这么多，其实下面的才是关键，针对于之前工作上面所面临的一些特有问题，以及团队内其他同事面临的一些挑战，我们做了一个基于spring aop的流控组件，特点是方便业务定制与配置，不需要代码嵌入，几乎没有依赖外部系统(依赖越多，系统的稳定性就会不断下降)。

图5 流控组件的实现

流控组件是基于spring aop实现的，因此，配置起来比较灵活和方便，流控锁实现的原理如上图，当请求进来时，调用配置的concurrentlock的enter方法，判断是否达到阀值，如果没有达到阀值，则进入，进行处理， 处理完后计数器加1，如果已经达到阀值则放入等待队列，因为等待队列是消耗内存的，因此等待队列也必须有阀值，如果队列超过阀值，请求直接丢弃。当然，实现也包括一些细节，具体请参看代码。

 

代码这里暂时就先不分享了吧，等有空把关键的那部分代码贴出来

评论

2 楼 lecheng1986 2017-07-22  

代码求分享下

1 楼 daysinsun 2015-09-23  

你这最后一段代码有问题吧，release方法里面你让count加了1，假如notify的object不为空，则你这个应该让count减1的。当有线程调用enter方法那就可以直接运行下去了，并不会wait，所以你这个没真正做到流控。