Semaphore使用与实现原理详解(看这篇就够了)

Semaphore使用与实现原理详解(看这篇就够了)-mikechen

Java并发编程提供了并发工具类，Semaphore就是其中典型的并发代表，今天重点详解Semaphore使用以及内部实现机制。

什么是Semaphore

Semaphore(信号量)，是JUC包下的一个工具类，我们可以通过其限制执行的线程数量，达到限流的效果。

当一个线程执行时先通过其方法进行获取许可操作，获取到许可的线程继续执行业务逻辑，当线程执行完成后进行释放许可操作，未获取达到许可的线程进行等待或者直接结束。

 

Semaphore的使用场景

Semaphore可以用来做流量分流，特别是对公共资源有限的场景，比如数据库连接。

假设有这个的需求，读取几万个文件的数据到数据库中，由于文件读取是IO密集型任务，可以启动几十个线程并发读取，但是数据库连接数只有10个，这时就必须控制最多只有10个线程能够拿到数据库连接进行操作。这个时候，就可以使用Semaphore做流量控制。

ublic class SemaphoreTest {
    private static final int COUNT = 40;
    private static Executor executor = Executors.newFixedThreadPool(COUNT);
    private static Semaphore semaphore = new Semaphore(10);
    public static void main(String[] args) {
        for (int i=0; i< COUNT; i++) {
            executor.execute(new ThreadTest.Task());
        }
    }

    static class Task implements Runnable {
        @Override
        public void run() {
            try {
                //读取文件操作
                semaphore.acquire();
                // 存数据过程
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
            }
        }
    }
}

 

Semaphore基本使用

Semaphore的使用也是比较简单的，我们创建一个Runnable的子类，如下：

private static class MyRunnable implements Runnable {
    // 成员属性 Semaphore对象
    private final Semaphore semaphore;


    public MyRunnable(Semaphore semaphore) {
        this.semaphore = semaphore;
    }


    public void run() {
        String threadName = Thread.currentThread().getName();
        // 获取许可
        boolean acquire = semaphore.tryAcquire();
        // 未获取到许可 结束
        if (!acquire) {
            System.out.println("线程【" + threadName + "】未获取到许可，结束");
            return;
        }
    // 获取到许可
        try {
            System.out.println("线程【" + threadName + "】获取到许可");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 释放许可
            semaphore.release();
            System.out.println("线程【" + threadName + "】释放许可");
        }
    }
}

测试方法如下：

public static void main(String[] args) {
    Semaphore semaphore = new Semaphore(2);
    for (int i = 0; i <= 10; i ++) {
        MyRunnable runnable = new MyRunnable(semaphore);
        Thread thread = new Thread(runnable, "Thread-" + i);
        thread.start();
    }
}

执行结果如下

Semaphore使用与实现原理详解(看这篇就够了)-mikechen

Semaphore实现原理

Semaphore的类图如下图所示：

Semaphore使用与实现原理详解(看这篇就够了)-mikechen

Sync是Semaphore的一个内部类，该类继承AQS，这个类又有公平和非公平的两个子类，这个内置的同步器实现Semaphore的功能。

Semaphore构造方法

在Semaphore中提供了两个构造方法，如下：

// 指定许可数量
public Semaphore(int permits) {
    // sync属性赋值   默认未非公平实现
    sync = new NonfairSync(permits);
}


// 指定许可数量和是否公平实现
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

内置同步器的构造方法如下：

FairSync(int permits) {
    super(permits);
}


NonfairSync(int permits) {
    super(permits);
}


Sync(int permits) {
    // 用state属性记录许可量
    setState(permits);
}

这些构造方法的逻辑是比较简单的哈，相信大家还记得在AQS中有一个state属性，当创建Semaphore时会将传递过来的许可量设置到同步器的state值，并将创建的同步器对象赋值给Semaphore中的sync属性。

获取许可

在Semaphore中提供了如下的获取许可的方法

  • void acquire() throws InterruptedException 获取一个许可，会阻塞等待其他线程释放许可
  • void acquire(int permits) throws InterruptedException 获取指定的许可数 ，会阻塞等待其他线程释放
  • void acquireUninterruptibly() 获取一个许可 会阻塞等待其他线程释放许可 可被中断
  • void acquireUninterruptibly(int permits) 获取指定的许可数 会阻塞等待其他线程释放许可 可被中断
  • boolean tryAcquire() 尝试获取许可 不会进行阻塞等待
  • boolean tryAcquire(int permits) 尝试获取指定的许可数 不会阻塞等待
  • boolean tryAcquire(long timeout, TimeUnit unit) 尝试获取许可 可指定等待时间
  • boolean tryAcquire(int permits, long timeout, TimeUnit unit) 尝试获取指定的许可数 可指定等待时间

由于篇幅有限，我们这里不全部介绍了，我们只介绍下acquire()和tryAcquire()两个方法，其他的方法实现大家自行查看吧，相差也不是很大。

 tryAcquire方法

这个方法的返回值表示是否获取许可成功，不会阻塞等待其他线程释放许可，没有许可了会直接返回false，其源码如下：

public boolean tryAcquire() {
    // 调用Semaphore.Sync中的nonfairTryAcquireShared方法
    return sync.nonfairTryAcquireShared(1) >= 0;
}


final int nonfairTryAcquireShared(int acquires) {
    // 自旋
    for (;;) {
        // 获取剩余的许可量
        int available = getState();
        // 扣减需要的信号量后的值
        int remaining = available - acquires;
        // 信号量不足 或者CAS替换state失败  返回扣减后的信号量值
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

这个方法的逻辑比较简单，判断许可量是否充足，充足的话CAS修改state的值。判断分配所需数量后的值是否大于等于0。

 acquire方法

这个方法没有返回值，当许可不足时会阻塞线程等待其他线程释放许可，其源码如下：

public void acquire() throws InterruptedException {
    // 调用AQS中的方法
    sync.acquireSharedInterruptibly(1);
}


// AQS中的方法
public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // tryAcquireShared是个模板方法，需要子类去实现
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

在acquire的方法中，会调用到AQS中的acquireSharedInterruptibly方法，在这个方法中用到了模板方法模式，tryAcquireShared方法是一个模板方法，需要子类去实现。接下来我们分别看看在Semaphore中的公平和非公平模式都是如何实现的。

公平模式下的tryAcquireShared方法实现如下：

protected int tryAcquireShared(int acquires) {
    // 自旋
    for (;;) {
        // 判断是否已经存在阻塞的线程
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

非公平模式下的tryAcquireShared方法实现如下：

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}


final int nonfairTryAcquireShared(int acquires) {
    // 自旋
    for (;;) {
        // 获取剩余的许可量
        int available = getState();
        // 扣减需要的信号量后的值
        int remaining = available - acquires;
        // 信号量不足 或者CAS替换state失败  返回扣减后的信号量值
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

公平和非公平模式的实现的区别是在公平模式的实现中会先判断是否已经存在阻塞的线程了，存在的话不会再去竞争获取许可了。

AQS.doAcquireSharedInterruptibly方法的逻辑如下：

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 添加到AQS的阻塞队列中
    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);
    }
}

释放许可

释放许可的方法为release方法，其源码如下：

public void release() {
    sync.releaseShared(1);
}


// AQS中的方法
public final boolean releaseShared(int arg) {
    // 模板方法  需要子类实现
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        // 成功
        return true;
    }
    // 失败
    return false;
}

Semaphore.Sync实现的tryReleaseShared方法逻辑如下：

protected final boolean tryReleaseShared(int releases) {
    // 自旋
    for (;;) {
        // 获取当前的许可量
        int current = getState();
        // 加上需要释放的量
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        // CAS修改state
        if (compareAndSetState(current, next))
            return true;
    }
}

AQS.doreleaseShared的逻辑如下：

private void doReleaseShared() {
    // 自旋
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // 唤醒下个节点  LockSupport.unpark
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

 

Semaphore总结

Semaphore主要用于控制当前活动线程数目，就如同停车场系统一般，而Semaphore则相当于看守的人，用于控制总共允许停车的停车位的个数，而对于每辆车来说就如同一个线程，线程需要通过acquire()方法获取许可，而release()释放许可。

如果许可数达到最大活动数，那么调用acquire()之后，便进入等待队列，等待已获得许可的线程释放许可，从而使得多线程能够合理的运行。

作者简介

陈睿|mikechen(个人微信：mikechen68)，10年+大厂架构经验，曾就职于阿里巴巴、淘宝、百度等一线互联网大厂。

以上!

关注作者「mikechen」公众号，获取更多架构面试干货！

后台回复【架构】，即可获取《阿里架构师进阶专题全部合集》，后台回复【面试】即可获取《史上最全阿里Java面试题总结》

相关推荐

史上最全阿里Java面试题总结

Java面试题答案总结(2023版)

2023最新大厂Java高频题全解

史上最全大厂Java面试题总结

相关面试

史上最全Java面试题答案总结

史上最全Spring面试题及答案

史上最全Redis面试题及答案

史上最全MySQL面试题及答案

评论交流
    说说你的看法
❯
返回顶部