如何使用Java高并发编程之Semaphore

本篇内容主要讲解“如何使用Java高并发编程之Semaphore”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“如何使用Java高并发编程之Semaphore”吧!

在安州等地区,都构建了全面的区域性战略布局,加强发展的系统性、市场前瞻性、产品创新能力,以专注、极致的服务理念,为客户提供成都网站设计、成都网站建设 网站设计制作按需网站设计,公司网站建设,企业网站建设,品牌网站设计,营销型网站,成都外贸网站建设,安州网站建设费用合理。

共享锁、独占锁

学习semaphore之前我们必须要先了解下什么是共享锁。

共享锁:它是允许多个线程同时获取锁,并发的访问共享资源

独占锁:也有人把它叫做“独享锁”,它是是独占的,排他的,只能被一个线程可持有,  当独占锁已经被某个线程持有时,其他线程只能等待它被释放后,才能去争锁,并且同一时刻只有一个线程能争锁成功。

什么是Semaphore

Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源。很多年以来,我都觉得从字面上很难理解Semaphore所表达的含义,只能把它比作是控制流量的红绿灯,比如XX马路要限制流量,只允许同时有一百辆车在这条路上行使,其他的都必须在路口等待,所以前一百辆车会看到绿灯,可以开进这条马路,后面的车会看到红灯,不能驶入XX马路,但是如果前一百辆中有五辆车已经离开了XX马路,那么后面就允许有5辆车驶入马路,这个例子里说的车就是线程,驶入马路就表示线程在执行,离开马路就表示线程执行完成,看见红灯就表示线程被阻塞,不能执行。”

  • Semaphore机制是提供给线程抢占式获取许可,所以他可以实现公平或者非公平,类似于ReentrantLock。说了这么多我们来个实际的例子看一看,比如我们去停车场停车,停车场总共只有5个车位,但是现在有8辆汽车来停车,剩下的3辆汽车要么等其他汽车开走后进行停车,或者去找别的停车位?

/**  * @author: 公众号【Java金融】  */ public class SemaphoreTest {     public static void main(String[] args) throws InterruptedException {          // 初始化五个车位         Semaphore semaphore = new Semaphore(5);         // 等所有车子         final CountDownLatch latch = new CountDownLatch(8);         for (int i = 0; i < 8; i++) {             int finalI = i;             if (i == 5) {                 Thread.sleep(1000);                 new Thread(() -> {                     stopCarNotWait(semaphore, finalI);                     latch.countDown();                 }).start();                 continue;             }             new Thread(() -> {                 stopCarWait(semaphore, finalI);                 latch.countDown();             }).start();         }         latch.await();         log("总共还剩:" + semaphore.availablePermits() + "个车位");     }      private static void stopCarWait(Semaphore semaphore, int finalI) {         String format = String.format("车牌号%d", finalI);         try {             semaphore.acquire(1);             log(format + "找到车位了,去停车了");             Thread.sleep(10000);         } catch (Exception e) {             e.printStackTrace();         } finally {             semaphore.release(1);             log(format + "开走了");         }     }      private static void stopCarNotWait(Semaphore semaphore, int finalI) {          String format = String.format("车牌号%d", finalI);         try {             if (semaphore.tryAcquire()) {                 log(format + "找到车位了,去停车了");                 Thread.sleep(10000);                 log(format + "开走了");                 semaphore.release();             } else {                 log(format + "没有停车位了,不在这里等了去其他地方停车去了");             }         } catch (Exception e) {             e.printStackTrace();         }      }      public static void log(String content) {         // 格式化         DateTimeFormatter fmTime = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");         // 当前时间         LocalDateTime now = LocalDateTime.now();         System.out.println(now.format(fmTime) + "  "+content);     } }
2021-03-01 18:54:57  车牌号0找到车位了,去停车了 2021-03-01 18:54:57  车牌号3找到车位了,去停车了 2021-03-01 18:54:57  车牌号2找到车位了,去停车了 2021-03-01 18:54:57  车牌号1找到车位了,去停车了 2021-03-01 18:54:57  车牌号4找到车位了,去停车了 2021-03-01 18:54:58  车牌号5没有停车位了,不在这里等了去其他地方停车去了 2021-03-01 18:55:07  车牌号7找到车位了,去停车了 2021-03-01 18:55:07  车牌号6找到车位了,去停车了 2021-03-01 18:55:07  车牌号2开走了 2021-03-01 18:55:07  车牌号0开走了 2021-03-01 18:55:07  车牌号3开走了 2021-03-01 18:55:07  车牌号4开走了 2021-03-01 18:55:07  车牌号1开走了 2021-03-01 18:55:17  车牌号7开走了 2021-03-01 18:55:17  车牌号6开走了 2021-03-01 18:55:17  总共还剩:5个车位

从输出结果我们可以看到车牌号5这辆车看见没有车位了,就不在这个地方傻傻的等了,而是去其他地方了,但是车牌号6和车牌号7分别需要等到车库开出两辆车空出两个车位后才停进去。这就体现了Semaphore  的acquire 方法如果没有获取到凭证它就会阻塞,而tryAcquire方法如果没有获取到凭证不会阻塞的。

semaphore在dubbo中的应用

在Dubbo中可以给Provider配置线程池大小来控制系统提供服务的最大并行度,默认是200。

比如我现在这个订单系统有三个接口,分别为创单、取消订单、修改订单。这三个接口加起来的并发是200但是创单接口是核心接口,我想让它多分点线程来执行  让它可以有最大150个线程,取消订单和修改订单分别最大25个线程执行就可以了。dubbo提供了executes这一属性来实现这个功能

  

我们可以看看dubbo内部是如何来executes的,具体实现是在ExecuteLimitFilter这个类我们可以

 public Result invoke(Invoker invoker, Invocation invocation) throws RpcException {         URL url = invoker.getUrl();         String methodName = invocation.getMethodName();         Semaphore executesLimit = null;         boolean acquireResult = false;         int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0);         if (max > 0) {             RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName());             // 如果当前使用的线程数量已经大于等于设置的阈值,那么直接抛出异常 //            if (count.getActive() >= max) { // throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service // using threads greater than  limited.");             /**              * http://manzhizhen.iteye.com/blog/2386408              * use semaphore for concurrency control (to limit thread number)              */                           executesLimit = count.getSemaphore(max);             if(executesLimit != null && !(acquireResult = executesLimit.tryAcquire())) {                 throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than  limited.");             }         }         long begin = System.currentTimeMillis();         boolean isSuccess = true;         // 计数器+1         RpcStatus.beginCount(url, methodName);         try {             Result result = invoker.invoke(invocation);             return result;         } catch (Throwable t) {             isSuccess = false;             if (t instanceof RuntimeException) {                 throw (RuntimeException) t;             } else {                 throw new RpcException("unexpected exception when ExecuteLimitFilter", t);             }         } finally {            // 计数器-1             RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isSuccess);             if(acquireResult) {                 executesLimit.release();             }         }     }

从上述代码我们也可以看出早期这个是没有采用Semaphore来实现的,而是直接采用被注释的 if (count.getActive() >=  max) 这个来来实现的,由于这个count.getActive() >= max  和这个计数加1不是原子性的,所以会有问题,具体bug号可以看https://github.com/apache/dubbo/pull/582后面才采用上述代码用Semaphore来修复非原子性问题。具体更详细的分析可以参见代码的链接。不过现在最新版本(2.7.9)我看是采用采用自旋加上和CAS来实现的。

Semaphore

上面就是对Semaphore一个简单的使用以及dubbo中用到的例子,说句实话Semaphore在工作中用的还是比较少的,不过面试又有可能会被问到,所以还是有必要来一起学习一下它。我们前面《Java高并发编程基础之AQS》通过ReentrantLock  一起学习了下AQS,其实Semaphore同样也是通过AQS来是实现的,我们可以一起来对照下独占锁的方法,基本上都是有方法一一相对应的。图片这里有两点稍微需要注意的地方:

如何使用Java高并发编程之Semaphore

  • 在独占锁模式中,我们只有在获取了独占锁的节点释放锁时,才会唤醒后继节点,因为独占锁只能被一个线程持有,如果它还没有被释放,就没有必要去唤醒它的后继节点。

  • 在共享锁模式下,当一个节点获取到了共享锁,我们在获取成功后就可以唤醒后继节点了,而不需要等到该节点释放锁的时候,这是因为共享锁可以被多个线程同时持有,一个锁获取到了,则后继的节点都可以直接来获取。因此,在共享锁模式下,在获取锁和释放锁结束时,都会唤醒后继节点。

获取凭证

我们同样还是通过非公平锁的模式来获取凭证 我们可以看下acquire的核心方法

public final void acquireSharedInterruptibly(int arg)           throws InterruptedException {       if (Thread.interrupted())           throw new InterruptedException();       if (tryAcquireShared(arg) < 0)           doAcquireSharedInterruptibly(arg);   }    protected int tryAcquireShared(int acquires) {            return nonfairTryAcquireShared(acquires);   }  // 主要看下这个方法,这个方法返回的值也就是tryAcquireShared返回的值,因为tryAcquireShared->nonfairTryAcquireShared    final int nonfairTryAcquireShared(int acquires) {          //自旋    for (;;) {         //Semaphore用AQS的state变量的值代表可用许可数         int available = getState();         //可用许可数减去本次需要获取的许可数即为剩余许可数         int remaining = available - acquires;         //如果剩余许可数小于0或者CAS将当前可用许可数设置为剩余许可数成功,则返回成功许可数         if (remaining < 0 ||             compareAndSetState(available, remaining))             return remaining;     }
  • 当tryAcquireShared  获取返回许可书小于0时说明获取许可失败需要进入doAcquireSharedInterruptibly这个方法去休眠。

  • 当tryAcquireShared 获取返回许可书小于0时说明获取许可成功直接结束。

doAcquireSharedInterruptibly

private void doAcquireSharedInterruptibly(int arg)        throws InterruptedException {        // 独占锁的acquireQueued调用的是addWaiter(Node.EXCLUSIVE),        // 而共享锁调用的是addWaiter(Node.SHARED),表明了该节点处于共享模式        final Node node = addWaiter(Node.SHARED);        boolean failed = true;        try {            for (;;) {                final Node p = node.predecessor();                if (p == head) {                    int r = tryAcquireShared(arg);                    if (r >= 0) {                        setHeadAndPropagate(node, r);                        p.next = null; // help GC                        failed = false;                        return;                    }                }                if (shouldParkAfterFailedAcquire(p, node) &&                    parkAndCheckInterrupt())                    throw new InterruptedException();            }        } finally {            if (failed)                cancelAcquire(node);        }    }

这个方法是不是跟我们上篇文章讲的AQS的独占锁的acquireQueued很像,不过独占锁它是直接调用了用了setHead(node)方法,而共享锁调用的是setHeadAndPropagate(node,  r)这个方法除了调用setHead 里面还调用了doReleaseShared(唤醒后继节点)

private void setHeadAndPropagate(Node node, int propagate) {       Node h = head; // Record old head for check below       setHead(node);       if (propagate > 0 || h == null || h.waitStatus < 0 ||           (h = head) == null || h.waitStatus < 0) {           Node s = node.next;           if (s == null || s.isShared())               doReleaseShared();       }   }

其他的方法基本上是和ReentrantLock来实现的独占锁差不多,我相信大家对源码分析感兴趣的应该也不多,其他更多细节问题还是需要自己亲自动手去看源码的。

总结

当信号量Semaphore初始化设置许可证为1  时,它也可以当作互斥锁使用。其中0、1就相当于它的状态,当=1时表示其他线程可以获取,当=0时,排他,即其他线程必须要等待。

Semaphore是JUC包中的一个很简单的工具类,用来实现多线程下对于资源的同一时刻的访问线程数限制

Semaphore中存在一个【许可】的概念,即访问资源之前,先要获得许可,如果当前许可数量为0,那么线程阻塞,直到获得许可

Semaphore内部使用AQS实现,由抽象内部类Sync继承了AQS。因为Semaphore天生就是共享的场景,所以其内部实际上类似于共享锁的实现

共享锁的调用框架和独占锁很相似,它们最大的不同在于获取锁的逻辑——共享锁可以被多个线程同时持有,而独占锁同一时刻只能被一个线程持有。

由于共享锁同一时刻可以被多个线程持有,因此当头节点获取到共享锁时,可以立即唤醒后继节点来争锁,而不必等到释放锁的时候。因此,共享锁触发唤醒后继节点的行为可能有两处,一处在当前节点成功获得共享锁后,一处在当前节点释放共享锁后。

采用semaphore来进行限流的话会产生突刺现象。

★指在一定时间内的一小段时间内就用完了所有资源,后大部分时间中无资源可用。比如在限流方法中的计算器算法,设置1s内的最大请求数为100,在前100ms已经有了100个请求,则后面900ms将无法处理请求,这就是突刺现象。

到此,相信大家对“如何使用Java高并发编程之Semaphore”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!


分享文章:如何使用Java高并发编程之Semaphore
标题链接:http://hbruida.cn/article/gesijh.html