Semaphore源码简析

男娘i 2023-03-12 09:12 20阅读 0赞

之间已经讲过了CountdownLatch和CyclicBarrier，现在来讲讲Semaphore，Semaphore信号量的作用是指定个数的线程通过之后，主线程就能继续执行了，老规矩先上demo

public class SemaphoreTest { 
        public static void main(String[] args) throws Exception { 
            Semaphore semaphore = new Semaphore(0);
    
            new Thread() { 
                @Override
                public void run() { 
                    try { 
                        Thread.sleep(2000);
                        System.out.println("线程1执行完毕");
                        semaphore.release();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }.start();
    
            new Thread() { 
                @Override
                public void run() { 
                    try { 
                        Thread.sleep(1000);
                        System.out.println("线程2执行完毕");
                        semaphore.release();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }.start();
    
            semaphore.acquire(1);
            System.out.println("一个线程执行完毕");
        }
    }

输出

线程2执行完毕
    一个线程执行完毕
    线程1执行完毕

在demo中有三个地方使用到了Semaphore，一个是构造方法，一个是semaphore.acquire(1)，一个是semaphore.release()  
这三个方法，acquire方法就可以指定调用多少次release方法，当调用了指定数量的release方法之后，主线程才能继续往下执行，而且之后的线程也不会继续被阻塞，接下来分别分析一下源码中的这几个方法

首先是构造方法

// 这里的构造函数可以传两个参数，第二个参数指定非公平还是公平，不传的话默认是非公平的
    public Semaphore(int permits) { 
        sync = new NonfairSync(permits);
    }
    
    static final class NonfairSync extends Sync { 
        private static final long serialVersionUID = -2694183684443567898L;
    
        NonfairSync(int permits) { 
            // Sync(int permits) { 
            // setState(permits);
            // } 
            // 父类的方法我贴在上面了，这里Sync继承自AQS，将permits赋值给state，state在AQS中就是加锁的次数
            super(permits);
        }
    
        protected int tryAcquireShared(int acquires) { 
            return nonfairTryAcquireShared(acquires);
        }
    }

可以看到构造函数其实就是给state变量赋了个值，在demo的代码中，我们传的是0，所以这里state = 0  
我们接着看Semaphore的acquire方法

public void acquire(int permits) throws InterruptedException { 
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }
    
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException { 
        if (Thread.interrupted())
            throw new InterruptedException();
       	// tryAcquireShared其实调用的是Semaphore中的nonfairTryAcquireShared方法，具体在下面分析了
        if (tryAcquireShared(arg) < 0)
        	// 这个方法我们放到后面分析，这里我们先知道会在这里将线程挂起即可
            doAcquireSharedInterruptibly(arg);
    }
    
    final int nonfairTryAcquireShared(int acquires) { 
        for (;;) { 
        	// 在上面的构造函数中，我们设置了state = 0，所以这里的available = 0
            int available = getState();
            // remaining = 0 - 1 = -1
            int remaining = available - acquires;
            // remaining < 0则直接返回remaining，这里直接返回-1
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

之前说过了，acquire方法（假设参数为n）需要调用acquire方法参数的个数的release方法才能继续往下执行，demo中的代码，当两个工作线程还没执行的时候，主线程执行到acquire方法时，此时还没有其他线程调用release方法的时候，主线程被挂起来了

接下来是Semaphore的release方法

public void release() { 
        sync.releaseShared(1);
    }
    
    public final boolean releaseShared(int arg) { 
    	// 详细分析在下面，最后实现的效果就是 state += arg
        if (tryReleaseShared(arg)) { 
            // 唤醒阻塞的线程
            doReleaseShared();
            return true;
        }
        return false;
    }
    
    protected final boolean tryReleaseShared(int releases) { 
        for (;;) { 
            // 获取state，此时state还是0
            int current = getState();
            // releases入参为1，next = 1
            int next = current + releases;
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            // cas操作，将state设置为了next的值，此时就相当于state += releases
            if (compareAndSetState(current, next))
                return true;
        }
    }

分析完了release方法，结合之前的注释，我们知道了acquire方法在没有调用release方法的时候，其实主线程会被挂起，当调用了release方法之后，就会尝试唤醒阻塞的线程  
接下来我们再回过头来看看线程被挂起的那段逻辑，也就是之前acquire方法里面没有分析的doAcquireSharedInterruptibly方法

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) { 
                    // 还是执行上面分析过的nonfairTryAcquireShared方法
                    // 这里执行的核心就是用 state - arg（这个arg其实就是acquire方法的参数），然后返回结果
                    int r = tryAcquireShared(arg);
                    // r >= 0时进入，将所有阻塞住的线程唤醒继续执行
                    // 满足semaphore需要通过的线程数时，之后的线程就可以直接执行，不会被阻塞了
                    if (r >= 0) { 
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                // 最开始执行acquire方法时，就会进入下面的逻辑判断，使用LockSupport.park方法挂起当前线程
                // 如果在这被阻塞的线程被唤醒了，那么就会继续进入循环，重新判断Semaphore是否还会被阻塞
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

其实我也不知道我上面写了些什么玩意儿，如果看不懂，那就直接记结论吧

做个比喻来说明一下Semaphore的作用，有点类似于生产者消费者，Semaphore初始化的时候，相当于初始化库存量，当使用acquire方法的时候，就是来消费商品，当库存不足时，就会被阻塞住，而release方法，就是补充库存，当库存满足消费量时，才能继续往下执行

还有一种常见的用法就是在初始化构造的时候指定Semaphore的值，可以认为是一个计数器，然后调用acquire方法就相当于计数器减1，当计数器减为0之后，那么其他的线程调用acquire方法就会被挂起，直到一个调用acquire方法的线程执行release方法，被挂起的线程才能继续尝试调用acquire方法。其实可以类比一下线程池的用法，都可以限制工作的线程数量，hystrix的限流降级中就支持线程池和Semaphore两种方式

最后我们来梳理一下demo的运行流程吧，首先主线程构造了一个Semaphore，参数传的是0（相当于库存为0），接着开启两个工作线程，并且休眠一段时间，主线程执行了acquire方法，但是此时没有线程执行release方法，所以主线程被阻塞了（想消费，但是没有库存，只能等待补货了），隔了1s之后，线程2执行release方法（补货），输出 线程2执行完毕，那么此时主线程就会从阻塞中被唤醒（有足够的货物能够消费了）， 接着输出了 一个线程执行完毕，最后在系统执行2s之后，线程1执行release方法（继续补货），此时执行输出 线程1执行完毕