Java并发包源码学习系列：同步组件Semaphore源码解析

时间  2021-02-22
标签 java 编程 并发 ide 工具 学习 .net 线程 code 栏目 Java 繁體版
原文   原文链接

目录

Semaphore概述及案例学习

Semaphore信号量用来控制同时访问特定资源的线程数量，它经过协调各个线程，以保证合理地使用公共资源。java

public class SemaphoreTest {

    private static final int THREAD_COUNT = 30;
    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);

    private static Semaphore s = new Semaphore(10); //10个许可证数量,最大并发数为10

    public static void main(String[] args) {
        for(int i = 0; i < THREAD_COUNT; i ++){ //执行30个线程
            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    s.tryAcquire(); //尝试获取一个许可证
                    System.out.println("save data");
                    s.release(); //使用完以后归还许可证
                }
            });
        }
        threadPool.shutdown();
    }
}
  • 建立一个大小为30的线程池，可是信号量规定在10，保证许可证数量为10。
  • 每次线程调用tryAcquire()或者acquire()方法都会原子性的递减许可证的数量，release()会原子性递增许可证数量。

类图结构及重要字段

public class Semaphore implements java.io.Serializable {
    private static final long serialVersionUID = -3222578661600680210L;
    /** All mechanics via AbstractQueuedSynchronizer subclass */
    private final Sync sync;
    
    abstract static class Sync extends AbstractQueuedSynchronizer {
        // permits指定初始化信号量个数
        Sync(int permits) {
            setState(permits);
        }
        // ...
    }
    
    static final class NonfairSync extends Sync {...}
    
    static final class FairSync extends Sync {...}
    
    // 默认采用非公平策略
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
	
    // 能够指定公平策略
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }
    
    //...
}
  • 基于AQS，相似于ReentrantLock，Sync继承自AQS，有公平策略和非公平策略两种实现。
  • 相似于CountDownLatch，state在这里也是经过构造器指定，表示初始化信号量的个数。

本篇文章阅读须要创建在必定的AQS基础之上，这边推荐几篇前置文章，能够瞅一眼：编程

void acquire()

调用该方法时，表示但愿获取一个信号量资源，至关于acquire(1)。并发

若是当前信号量个数大于0，CAS将当前信号量值减1，成功后直接返回。ide

若是当前信号量个数等于0，则当前线程将被置入AQS的阻塞队列。工具

该方法是响应中断的，其余线程调用了该线程的interrupt()方法，将会抛出中断异常返回。学习

// Semaphore.java
	public void acquire() throws InterruptedException {
        // 传递的 arg 为 1 ， 获取1个信号量资源
        sync.acquireSharedInterruptibly(1);
    }
	// AQS.java
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 线程被 中断， 抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
        // 子类实现， 公平和非公平两种策略
        if (tryAcquireShared(arg) < 0)
            // 若是获取失败， 则置入阻塞队列， 
            // 再次进行尝试， 尝试失败则挂起当前线程
            doAcquireSharedInterruptibly(arg);
    }

非公平

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;

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

        protected int tryAcquireShared(int acquires) {
            // 这里直接调用Sync定义的 非公平共享模式获取方法
            return nonfairTryAcquireShared(acquires);
        }
    }

    abstract static class Sync extends AbstractQueuedSynchronizer {
        
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                // 获取当前信号量的值
                int available = getState();
                // 减去须要获取的值， 获得剩余的信号量个数
                int remaining = available - acquires;
                // 不剩了，表示当前信号量个数不能知足需求， 返回负数， 线程置入AQS阻塞
                // 还有的剩， CAS设置当前信号量值为剩余值， 并返回剩余值
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

你会发现，非公平策略是没法保证【AQS队列中阻塞的线程】和【当前线程】获取的顺序的，当前线程是有可能在排队的线程以前就拿到资源，产生插队现象。ui

公平策略就不同了，它会经过hasQueuedPredecessors()方法看看队列中是否存在前驱节点，以保证公平性。.net

公平策略

static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;

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

        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;
            }
        }
    }

void acquire(int permits)

在acquire()的基础上，指定了获取信号量的数量permits。线程

public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }

void acquireUninterruptibly()

该方法与acquire()相似，可是不响应中断。code

public void acquireUninterruptibly() {
        sync.acquireShared(1);
    }
	
    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

void acquireUninterruptibly(int permits)

该方法与acquire(permits)相似，可是不响应中断。

public void acquireUninterruptibly(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireShared(permits);
    }

boolean tryAcquire()

tryAcquire和acquire非公平策略公用一个逻辑，可是区别在于，若是获取信号量失败，或者CAS失败，将会直接返回false，而不会置入阻塞队列中。

通常try开头的方法的特色就是这样，尝试一下，成功是最好，失败也不至于被阻塞，而是马上返回false。

public boolean tryAcquire() {
        return sync.nonfairTryAcquireShared(1) >= 0;
    }
    abstract static class Sync extends AbstractQueuedSynchronizer {
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

boolean tryAcquire(int permits)

相比于普通的tryAcquire()，指定了permits的值。

public boolean tryAcquire(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.nonfairTryAcquireShared(permits) >= 0;
    }

boolean tryAcquire(int permits, long timeout, TimeUnit unit)

相比于tryAcquire(int permits)，增长了超时控制。

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
    }

void release()

将信号量值加1，若是有线程由于调用acquire方法而被阻塞在AQS阻塞队列中，将根据公平策略选择一个信号量个数知足需求的线程唤醒，线程唤醒后也会尝试获取新增的信号量。

参考文章：Java并发包源码学习系列：AQS共享模式获取与释放资源

// Semaphore.java
	public void release() {
        sync.releaseShared(1);
    }
	// AQS.java
    public final boolean releaseShared(int arg) {
        // 尝试释放锁
        if (tryReleaseShared(arg)) {
            // 释放成功， 唤醒AQS队列里面最早挂起的线程
            // https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112386838
            doReleaseShared();
            return true;
        }
        return false;
    }
	// Semaphore#Sync.java
    abstract static class Sync extends AbstractQueuedSynchronizer {
        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                // 获取当前信号量
                int current = getState();
                // 指望加上releases
                int next = current + releases;
                if (next < current) // overflow
                    throw new Error("Maximum permit count exceeded");
                // CAS操做，更新
                if (compareAndSetState(current, next))
                    return true;
            }
        }
    }

void release(int permits)

和release()相比指定了permits的值。

public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }

其余方法

Semaphore还提供其余一些方法，实现比较简单，这边就简单写一下吧：

// 返回此信号量中当前可用的许可证数量， 其实就是获得当前的 state值  getState()
	public int availablePermits() {
        return sync.getPermits();
    }

    // 将state更新为0， 返回0
    public int drainPermits() {
        return sync.drainPermits();
    }

	// 减小reduction个许可证
    protected void reducePermits(int reduction) {
        if (reduction < 0) throw new IllegalArgumentException();
        sync.reducePermits(reduction);
    }

    // 判断公平策略
    public boolean isFair() {
        return sync instanceof FairSync;
    }

	// 判断是否有线程证在等待获取许可证
    public final boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }

	// 返回正在等待获取许可证的线程数
    public final int getQueueLength() {
        return sync.getQueueLength();
    }

	// 返回全部等待获取许可证的线程集合
    protected Collection<Thread> getQueuedThreads() {
        return sync.getQueuedThreads();
    }

总结

Semaphore信号量用来控制同时访问特定资源的线程数量，它经过协调各个线程，以保证合理地使用公共资源。

  • 基于AQS，相似于ReentrantLock，Sync继承自AQS，有公平策略和非公平策略两种实现。
  • 相似于CountDownLatch，state在这里也是经过构造器指定，表示初始化信号量的个数。

每次线程调用tryAcquire()或者acquire()方法都会原子性的递减许可证的数量，release()会原子性递增许可证数量，只要有许可证就能够重复使用。

参考阅读

  • 《Java并发编程之美》
  • 《Java并发编程的艺术》
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程