Semaphore信号量的使用与源码分析、使用wait-notify手写实现

朱雀 2022-11-03 01:38 173阅读 0赞

Semaphore（信号量）是用来限制资源（如线程）的并发数量。

# api #

<table> 
 <thead> 
  <tr> 
   <th>方法名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>void acquire()</td> 
   <td>从信号量获取1个许可证，一直阻塞到有许可证或被中断</td> 
  </tr> 
  <tr> 
   <td>void acquire(int permits)</td> 
   <td>从信号量获取permits个许可证，一直阻塞到有许可证或被中断</td> 
  </tr> 
  <tr> 
   <td>void acquireUninterruptibly()</td> 
   <td>从信号量获取1个许可证，一直阻塞到有许可证，不可被中断</td> 
  </tr> 
  <tr> 
   <td>void acquireUninterruptibly(int permits)</td> 
   <td>从信号量获取permits个许可证，一直阻塞到有许可证，不可被中断</td> 
  </tr> 
  <tr> 
   <td>int availablePermits()</td> 
   <td>返回此信号量中当前可用的许可证数</td> 
  </tr> 
  <tr> 
   <td>int drainPermits()</td> 
   <td>获取并返回所有可立即获得的许可证</td> 
  </tr> 
  <tr> 
   <td>int getQueueLength()</td> 
   <td>返回等待获取许可证的线程数</td> 
  </tr> 
  <tr> 
   <td>boolean hasQueuedThreads()</td> 
   <td>返回是否有等待获取许可证的线程</td> 
  </tr> 
  <tr> 
   <td>boolean isFair()</td> 
   <td>如果此信号量的公平设置为真，则返回true，否则为false</td> 
  </tr> 
  <tr> 
   <td>void release()</td> 
   <td>释放许可证，将其返回到信号量</td> 
  </tr> 
  <tr> 
   <td>void release(int permits)</td> 
   <td>释放给定数量的许可证，将其返回到信号量</td> 
  </tr> 
  <tr> 
   <td>boolean tryAcquire()</td> 
   <td>从这个信号量尝试获得许可证，不阻塞</td> 
  </tr> 
  <tr> 
   <td>boolean tryAcquire(int permits)</td> 
   <td>从这个信号量尝试获得给定数量的许可证，不阻塞</td> 
  </tr> 
  <tr> 
   <td>boolean tryAcquire(int permits, long timeout, TimeUnit unit)</td> 
   <td>在给定的等待时间内从该信号量尝试获取给定数量的许可证</td> 
  </tr> 
  <tr> 
   <td>boolean tryAcquire(long timeout, TimeUnit unit)</td> 
   <td>在给定的等待时间内从该信号量尝试获取1个许可证</td> 
  </tr> 
 </tbody> 
</table>

# 简单使用 #

模拟场景：10个线程同时存储数据到数据库，而数据库的连接数只有3个，这时我们必须控制只有3个线程同时获取数据库连接保存数据，否则会报错无法获取数据库连接。

package com.morris.concurrent.tool.semaphore.api;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    public class SemaphoreDemo { 
    
        private static Semaphore semaphore = new Semaphore(3);
        private static ExecutorService executor = Executors.newFixedThreadPool(10);
    
        public static void main(String[] args) { 
    
            for (int i = 0; i < 10; i++) { 
                executor.execute(() -> { 
    
                    try { 
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + "当前等待的线程数：" + semaphore.getQueueLength());
                        System.out.println(Thread.currentThread().getName() + "当前可用的凭证数：" + semaphore.availablePermits());
                        System.out.println("save data");
                        TimeUnit.SECONDS.sleep(1);
                        semaphore.release();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
    
                });
            }
            executor.shutdown();
        }
    
    }

运行结果如下：

pool-1-thread-2当前等待的线程数：0
    pool-1-thread-2当前可用的凭证数：2
    save data
    pool-1-thread-3当前等待的线程数：0
    pool-1-thread-3当前可用的凭证数：1
    save data
    pool-1-thread-4当前等待的线程数：0
    pool-1-thread-4当前可用的凭证数：0
    save data
    pool-1-thread-7当前等待的线程数：6
    pool-1-thread-7当前可用的凭证数：1
    save data
    pool-1-thread-1当前等待的线程数：5
    pool-1-thread-1当前可用的凭证数：0
    save data
    pool-1-thread-5当前等待的线程数：4
    pool-1-thread-5当前可用的凭证数：0
    save data
    pool-1-thread-6当前等待的线程数：3
    pool-1-thread-8当前等待的线程数：2
    pool-1-thread-8当前可用的凭证数：0
    save data
    pool-1-thread-6当前可用的凭证数：0
    save data
    pool-1-thread-10当前等待的线程数：1
    pool-1-thread-10当前可用的凭证数：0
    save data
    pool-1-thread-9当前等待的线程数：0
    pool-1-thread-9当前可用的凭证数：1
    save data

# 综合使用 #

可见Semaphore的构造方法中的许可证个数只是一个初始值，可以通过release动态添加许可证。

package com.morris.concurrent.tool.semaphore.api;
    
    import java.util.concurrent.Semaphore;
    
    public class SemaphoreComplexDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            Semaphore semaphore = new Semaphore(10);
    
            semaphore.acquire(2); // 一次获取多个凭证
            semaphore.acquire(2);
            semaphore.acquire(2);
            semaphore.acquire(2);
            semaphore.acquire(2);
    
            System.out.println(semaphore.availablePermits());
    
            semaphore.release(10); // 动态添加凭证
            System.out.println(semaphore.availablePermits());
    
            semaphore.release(6);
            System.out.println(semaphore.availablePermits());
    
    
            semaphore.drainPermits(); // 获取所有的凭证
            System.out.println(semaphore.availablePermits());
        }
    }

# 总结 #

*  Semaphore并不能保证acquire()和release()中间代码的原子性，如果有共享资源的话，在高并发下可能会线程安全问题。
 *  所谓公平就是获取许可证的顺序与调用acquire的顺序有关，谁先调用acquire谁先获得许可证，但是不能百分之百保证，仅仅在概率上保证。

# 源码分析 #

## 构造方法 ##

java.util.concurrent.Semaphore\#Semaphore(int)

public Semaphore(int permits) { 
        sync = new NonfairSync(permits);
    }

java.util.concurrent.Semaphore.NonfairSync\#NonfairSync

static final class NonfairSync extends Sync { 
    
        NonfairSync(int permits) { 
            super(permits);
        }
    ... ...

java.util.concurrent.Semaphore.Sync\#Sync

abstract static class Sync extends AbstractQueuedSynchronizer { 
    
        Sync(int permits) { 
            setState(permits);
        }
    ... ...

Semaphore底层使用了AQS，并将AQS中state的初始值设置为permits，也就是state为资源的数量。

## acquire ##

java.util.concurrent.Semaphore\#acquire()

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

这里调用了AQS中获取共享状态的方法acquireSharedInterruptibly()，acquireSharedInterruptibly()中使用了模板方法模式，会调用子类Semaphore.NonfairSync.tryAcquireShared()。

java.util.concurrent.Semaphore.NonfairSync\#tryAcquireShared

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

java.util.concurrent.Semaphore.Sync\#nonfairTryAcquireShared

final int nonfairTryAcquireShared(int acquires) { 
        for (;;) { 
            int available = getState();
            int remaining = available - acquires;
     // 扣减获取的许可凭证书数
    		// remaining>=0表示成功，remaining < 0表示失败
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

当许可凭证数足够时，那么tryAcquireShared就会返回大于等于0，然后线程往下执行，当许可凭证数不足时，就会进入AQS中的同步队列中进行等待，直到被其他线程唤醒。

## release ##

java.util.concurrent.Semaphore\#release()

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

这里调用了AQS中释放共享状态的方法releaseShared()，releaseShared()中也使用了模板方法模式，会调用子类Semaphore.NonfairSync.tryReleaseShared()。

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

tryReleaseShared会释放许可凭证数量，然后唤醒同步队列中等待的线程。

# 手写实现 #

package com.morris.concurrent.tool.semaphore.my;
    
    import java.util.*;
    
    /** * 使用wait-notify实现Semaphore */
    public class WaitNotifySemaphore { 
    
        private int permits;
    
        private Set<Thread> waitThreads = new HashSet<>(); // 等待的线程数
    
        public WaitNotifySemaphore(int permits) { 
            this.permits = permits;
        }
    
        public synchronized void acquire() throws InterruptedException { 
            while (permits == 0) { 
                waitThreads.add(Thread.currentThread());
                this.wait();
            }
            waitThreads.remove(Thread.currentThread());
            permits--;
            this.notifyAll();
        }
    
        public synchronized void release() { 
            permits++;
            this.notifyAll();
        }
    
        public int availablePermits() { 
            return permits;
        }
    
        public boolean hasQueuedThreads() { 
            return waitThreads.isEmpty();
        }
    
        public int getQueueLength() { 
            return waitThreads.size();
        }
    
        protected Collection<Thread> getQueuedThreads() { 
            return Collections.unmodifiableSet(waitThreads);
        }
    
        public String toString() { 
            return super.toString() + "[Permits = " + permits + "]";
        }
    }