Semaphore1.8源码

逃离我推掉我的手 2022-04-25 01:24 186阅读 0赞

Semaphore 主要用于限量控制并发执行代码的工具类, 其内部通过 一个 permit 来进行定义并发执行的数量, 其实可以理解为一个 限制数量的 ReadLock 获取.  
限制可以访问某些资源（物理或逻辑的）线程数目

加锁限制只能有一个线程同时访问资源  
Semaphore限制多个线程同时访问资源

Semaphore 主要特点:  
1.Semaphore 方法的实现通过 Sync(AQS的继承类)代理来实现  
2.支持公平与非公平模式, 都是在AQS的子类里面进行, 主要区分在 tryAcquire 里面

使用

package com.study.demo;
    
    import java.util.Date;
    import java.util.concurrent.Semaphore;
    
    /**
     * Created by Administrator on 2019\3\2 0002.
     */
    public class Test {
    
        public static void main(String[] args) throws InterruptedException {
            Semaphore windows = new Semaphore(2);  // 声明2个窗口
    
            for (int i = 0; i < 3; i++) {
                new Thread() {
                    @Override
                    public void run() {
                        try {
                            System.out.println(Thread.currentThread().getName() + " 开始占窗口 " + new Date());
                            windows.acquire();  // 占用窗口
                            System.out.println(Thread.currentThread().getName() + " 开始买票 " + new Date());
                            sleep(2000);  // 睡2秒，模拟买票流程
                            System.out.println(Thread.currentThread().getName() + " 购票成功 " + new Date());
                            windows.release();  // 释放窗口
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }.start();
            }
    
        }
    }

Thread-2 开始占窗口 Tue Apr 30 11:32:41 CST 2019  
Thread-1 开始占窗口 Tue Apr 30 11:32:41 CST 2019  
Thread-2 开始买票 Tue Apr 30 11:32:41 CST 2019  
Thread-1 开始买票 Tue Apr 30 11:32:41 CST 2019  
Thread-0 开始占窗口 Tue Apr 30 11:32:41 CST 2019  
Thread-1 购票成功 Tue Apr 30 11:32:43 CST 2019  
Thread-0 开始买票 Tue Apr 30 11:32:43 CST 2019  
Thread-2 购票成功 Tue Apr 30 11:32:43 CST 2019  
Thread-0 购票成功 Tue Apr 30 11:32:45 CST 2019

## 源码分析 ##

public class Semaphore implements java.io.Serializable {
        private static final long serialVersionUID = -3222578661600680210L;
        /** All mechanics via AbstractQueuedSynchronizer subclass */
        private final Sync sync;

构造器

public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
    
        /**
         * NonFair version
         */
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = -2694183684443567898L;
    
            NonfairSync(int permits) {
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
                return nonfairTryAcquireShared(acquires);
            }
        }
    
        abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 1192457210091910933L;
    
            Sync(int permits) {
                setState(permits);
            }

permits许可证数量，赋值给aqs的state， state=2

## 获取许可证 ##

public void acquire() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }

AbstractQueuedSynchronizer

/**
         * Acquires in shared mode, aborting if interrupted.  Implemented
         * by first checking interrupt status, then invoking at least once
         * {@link #tryAcquireShared}, returning on success.  Otherwise the
         * thread is queued, possibly repeatedly blocking and unblocking,
         * invoking {@link #tryAcquireShared} until success or the thread
         * is interrupted.
         */
        public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
        }

Semaphore-NonfairSync

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

Semaphore-Sync

final int nonfairTryAcquireShared(int acquires) {
                for (;;) {
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }

拿走一个许可证，2-1=1，剩下一个，state=1，返回1

Thread-0 开始占窗口 Tue Apr 30 11:32:41 CST 2019  
Thread-1 购票成功 Tue Apr 30 11:32:43 CST 2019  
Thread-0 开始买票 Tue Apr 30 11:32:43 CST 2019

Thread-0尝试获取许可证，返回-1，调用doAcquireSharedInterruptibly，创建节点，共享模式，入队，设置对头节点状态signal，然后自己阻塞

/**
         * Acquires in shared interruptible mode.
         * @param arg the acquire argument
         */
        private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            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);
            }
        }

## 释放许可证 ##

/**
         * Releases a permit, returning it to the semaphore.
         *
         * <p>Releases a permit, increasing the number of available permits by
         * one.  If any threads are trying to acquire a permit, then one is
         * selected and given the permit that was just released.  That thread
         * is (re)enabled for thread scheduling purposes.
         *
         * <p>There is no requirement that a thread that releases a permit must
         * have acquired that permit by calling {@link #acquire}.
         * Correct usage of a semaphore is established by programming convention
         * in the application.
         */
        public void release() {
            sync.releaseShared(1);
        }

AbstractQueuedSynchronizer

/**
         * Releases in shared mode.  Implemented by unblocking one or more
         * threads if {@link #tryReleaseShared} returns true.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryReleaseShared} but is otherwise uninterpreted
         *        and can represent anything you like.
         * @return the value returned from {@link #tryReleaseShared}
         */
        public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }

Semaphore-Sync

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");
                    if (compareAndSetState(current, next))
                        return true;
                }
            }

把state+1

AbstractQueuedSynchronizer

/**
         * Release action for shared mode -- signals successor and ensures
         * propagation. (Note: For exclusive mode, release just amounts
         * to calling unparkSuccessor of head if it needs signal.)
         */
        private void doReleaseShared() {
            /*
             * Ensure that a release propagates, even if there are other
             * in-progress acquires/releases.  This proceeds in the usual
             * way of trying to unparkSuccessor of head if it needs
             * signal. But if it does not, status is set to PROPAGATE to
             * ensure that upon release, propagation continues.
             * Additionally, we must loop in case a new node is added
             * while we are doing this. Also, unlike other uses of
             * unparkSuccessor, we need to know if CAS to reset status
             * fails, if so rechecking.
             */
            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
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

如果头结点=null，直接结束。如果头结点部位null，说明有等待线程，唤醒他