互联网技术14——Semaphore信号量

SemaPhore信号量非常适合高并发访问，在系统上线之前，要对系统访问量记性评估，当然这个肯定不是随便写的，是经过以往的经验、数据、历年的访问量，已经推广力度进行一个合理的评估，当然评估标准不能太大也不能太小。太大的话则投入的资源得不到实际的效果，会浪费资源。当评估值调小则某个时间点一个高峰值的访问量上回直接压垮系统。

解决高并发关键在业务，要将业务进行划分，划分为不同的层次和模块，针对某个具体的模块，使用Nginx等进行分流，分给部署同一个系统的多个Tomcat服务器，使其负载均衡。整个架构可能包括商品系统。促销系统、购物车系统。支付系统等，在网络入口处视同nginx，将流量分配到不同的模块，分别进行处理。下面是互联网企业关系的一些相关概念，Nginx理论上可以大概处理2000万的并发类，若并发量更大，可以考虑在Nginx之前，添加LVS。

pv（page view）：网站的总访问量，页面浏览量或点击量，用户每刷新一次就会被记录一次。

UV(unique visitor)：访问网站的一台电脑客户端为一个方可，一般来讲，时间上以00.00-24“00之间相同的id的客户端只记录一次。

QPB（query per second）：每秒查询书，qps很大程度上代表了系统业务上的繁忙程度，每次请求的背后，可能对应着多次磁盘io、多次忘了请求，多个cpu时间片等。一旦当前qps超过锁设定的预警阈值，可以考虑增机器对集群扩容，一面压力过大导致宕机，可以根据前期的压力测试得到估值，在结合后期综合运维的情况，估算出阈值。

RT（response time）请求的相应时间，这个指标非常关键，直接说明前段用户的体验。因此任何系统设计师动想降低rt的时间。

当然还涉及cpu、内存、网络、磁盘等情况，更多细节的问题很多，如：每秒select、update、delete操作的数据量扥数据层面的统计。

一般来讲，主要上线的系统进行多轮压力测试以后，可以对系统进行峰值评估，采用所谓的80/20原则，即80%的访问请求将在20%的时间内到达。这样我们可以根据系统对应的pv计算出峰值qps。

峰值qps=（总PV*80%）/(60*60*24*20%)

然后再讲总峰值qps除以单台机器所能承受的最高qps值，就是所需要机器的数量。

机器数=总得峰值qps/压力测试得到的单台机器极限qps

 

Semaphore控制系统的流量

semaphore又称信号量，是操作系统中的一个概念，在java并发编程中，信号量控制的是线程并发的数量。samephore管理一些列许可证。release释放一个许可证。acquire（）：当前行代码从samephore中获得一个许可证，release（）：当前行代码向samephore中释放一个许可证。

semaphire经常用于限制获取某种资源的线程数量。比如说有5条铁路，每个铁路上只能走一条火车，那们就可以初始化一个new  Semaphore（5），每次火车行驶前都要执行acquire（）来获取同行许可，如果当前行驶的火车数小于5则可直接获得许可，如果等于5，则要等待有一个火车行驶完毕，执行release（）归还许可。

示例代码：

我直接用的随机数表示是同一个线程，虽然不严瑾，但是在1000以内随机，不同线程得到相同随机数概率很小，作为演示，还是可以接受。

package com.company.concurrentTest;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * Created by BaiTianShi on 2018/8/23.
 */
public class SemaphoreTest {

    public static void main(String[] args) {
         Semaphore semaphore = new Semaphore(1);
         ExecutorService executorService = Executors.newCachedThreadPool();
        for(  int i=0;i<6;i++){
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        int x=(int)(Math.random()*1000);
                        semaphore.acquire();
                        System.out.println("我产生的随机数是"+x+"，开始执行任务");
                        Thread.sleep(1000);
                        System.out.println("我产生的随机数是"+x+"，任务执行结束");
                        Thread.sleep(1000);
                        semaphore.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }



    }

}

运行结果：永远都是一个线程执行完毕，在执行另一个线程。

我是随机98线程开始
我是随机98线程结束
我是随机977线程开始
我是随机977线程结束
我是随机173线程开始
我是随机173线程结束
我是随机503线程开始
我是随机503线程结束
我是随机422线程开始
我是随机422线程结束
我是随机764线程开始
我是随机764线程结束

把new Semaphore(1) 改成 new Semaphore(3)，再看运行结果

可以看见，此时是前三个线程同时执行，然后一个线程执行完毕后释放一个信号，再开始一个等待着的线程。

同时samephore的acquire()方法和release()可以传入一个int类型的参数，用以指定一次性获取或释放许可证的数量。

同时samephore()还有两个两个重要的方法

1.avaliablePermits()：返回可用许可数量

2.drainPermits()：返回所有可用的许可数量，并将许可数置0

avaliablePermits示例：

package com.company.concurrentTest;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * Created by BaiTianShi on 2018/8/23.
 */
public class SemaphoreTest2 {

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        try {
            semaphore.acquire(2);
            System.out.println(semaphore.availablePermits());
            semaphore.release();
            System.out.println(semaphore.availablePermits());
            semaphore.release(5);
            System.out.println(semaphore.availablePermits());

        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

}

运行结果：

drainPermits（）使用示例：

package com.company.concurrentTest;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * Created by BaiTianShi on 2018/8/23.
 */
public class SemaphoreTest2 {

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        try {
            semaphore.acquire(2);
            //期望结果 1
            System.out.println(semaphore.availablePermits());
            semaphore.release();
            //期望结果 2
            System.out.println(semaphore.availablePermits());
            semaphore.release(5);
            //期望结果 7
            System.out.println(semaphore.availablePermits());

            //期望结果 7
            System.out.println(semaphore.drainPermits());
            //期望结果 0
            System.out.println(semaphore.availablePermits());

        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

}

运行结果：