Java 信号量 Semaphore 入门介绍

Z时代
2024-01-10
分类：综合

java

一、简介
二、概念
　　2.1、Semaphore信号量模型
　　2.2、Semaphore分为单值和多值两种，前者只能被一个线程获得，后者可以被若干个线程获得
　　2.3、公平/非公平模式
　　2.4、主要的方法
三、Semaphore应用场景
　　示例-1：Semaphore可以做到一个deadlock recovery的示例
　　示例2-Semaphore限流

一、简介

信号量(Semaphore)，有时被称为信号灯，是在多线程环境下使用的一种设施, 它负责协调各个线程, 以保证它们能够正确、合理的使用公共资源。

一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。拿到信号量的线程可以进入代码，否则就等待。通过acquire()和release()获取和释放访问许可。

Semaphore通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。

二、概念

2.1、Semaphore信号量模型

　　Semaphore（信号量）并不是 Java 语言特有的，几乎所有的并发语言都有。所以也就存在一个信号量模型的概念，如下图所示：

信号量模型比较简单，可以概括为：一个计数器、一个队列、三个方法。

计数器：记录当前还可以运行多少个线程访问资源。

队列：待访问资源的线程

三个方法：

init()：初始化计数器的值，可就是允许多少线程同时访问资源。

up()：计数器加1，有线程归还资源时，如果计数器的值大于或者等于 0 时，从等待队列中唤醒一个线程

down()：计数器减 1，有线程占用资源时，如果此时计数器的值小于 0 ，线程将被阻塞。

这三个方法都是原子性的，由实现方保证原子性。例如在 Java 语言中，JUC 包下的 Semaphore 实现了信号量模型，所以 Semaphore 保证了这三个方法的原子性。

2.2、Semaphore分为单值和多值两种，前者只能被一个线程获得，后者可以被若干个线程获得

将信号量初始化为 1，使得它在使用时最多只有一个可用的许可，从而可用作一个相互排斥的锁。这通常也称为二进制信号量，因为它只能有两种状态：一个可用的许可，或零个可用的许可。按此方式使用时，二进制信号量具有某种属性（与很多 Lock 实现不同），即可以由线程释放“锁”，而不是由所有者（因为信号量没有所有权的概念）。在某些专门的上下文（如死锁恢复）中这会很有用。

此类的构造方法可选地接受一个公平参数。当设置为 false 时，此类不对线程获取许可的顺序做任何保证。特别地，闯入是允许的，也就是说可以在已经等待的线程前为调用 acquire() 的线程分配一个许可，从逻辑上说，就是新线程将自己置于等待线程队列的头部。当公平设置为 true 时，信号量保证对于任何调用获取方法的线程而言，都按照处理它们调用这些方法的顺序（即先进先出；FIFO）来选择线程、获得许可。注意，FIFO 排序必然应用到这些方法内的指定内部执行点。所以，可能某个线程先于另一个线程调用了 acquire，但是却在该线程之后到达排序点，并且从方法返回时也类似。还要注意，非同步的 tryAcquire 方法不使用公平设置，而是使用任意可用的许可。

通常，应该将用于控制资源访问的信号量初始化为公平的，以确保所有线程都可访问资源。为其他的种类的同步控制使用信号量时，非公平排序的吞吐量优势通常要比公平考虑更为重要。

此类还提供便捷的方法来同时 acquire 和释放多个许可。小心，在未将公平设置为 true 时使用这些方法会增加不确定延期的风险。

内存一致性效果：线程中调用“释放”方法（比如 release()）之前的操作 happen-before 另一线程中紧跟在成功的“获取”方法（比如 acquire()）之后的操作。

2.3、公平/非公平模式

Semaphore 类中，实现了两种信号量：公平的信号量和非公平的信号量：

公平的信号量就是大家排好队，先到先进，

非公平的信号量就是不一定先到先进，允许插队。非公平的信号量效率会高一些，所以默认使用的是非公平信号量。

JDK中定义如下：
Semaphore(int permits, boolean fair)
　　创建具有给定的许可数和给定的公平设置的Semaphore。

2.4、主要的方法

构造方法摘要

Semaphore(int permits)
创建具有给定的许可数和非公平的公平设置的 Semaphore。

Semaphore(int permits, boolean fair)

创建具有给定的许可数和给定的公平设置的

方法摘要
`void`	`acquire()` 从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
`void`	`acquire(int permits)` 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞，或者线程已被中断。
`void`	`acquireUninterruptibly()` 从此信号量中获取许可，在有可用的许可前将其阻塞。
`void`	`acquireUninterruptibly(int permits)` 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞。
`int`	`availablePermits()` 返回此信号量中当前可用的许可数。
`int`	`drainPermits()` 获取并返回立即可用的所有许可。
`protected Collection<Thread>`	`getQueuedThreads()` 返回一个 collection，包含可能等待获取的线程。
`int`	`getQueueLength()` 返回正在等待获取的线程的估计数目。
`boolean`	`hasQueuedThreads()` 查询是否有线程正在等待获取。
`boolean`	`isFair()` 如果此信号量的公平设置为 true，则返回 `true`。
`protected` `void`	`reducePermits(int reduction)` 根据指定的缩减量减小可用许可的数目。
`void`	`release()` 释放一个许可，将其返回给信号量。
`void`	`release(int permits)` 释放给定数目的许可，将其返回到信号量。
`String`	`toString()` 返回标识此信号量的字符串，以及信号量的状态。
`boolean`	`tryAcquire()` 仅在调用时此信号量存在一个可用许可，才从信号量获取许可。
`boolean`	`tryAcquire(int permits)` 仅在调用时此信号量中有给定数目的许可时，才从此信号量中获取这些许可。
`boolean`	`tryAcquire(int permits,` `long timeout, TimeUnit unit)` 如果在给定的等待时间内此信号量有可用的所有许可，并且当前线程未被中断，则从此信号量获取给定数目的许可。
`boolean`	`tryAcquire(long timeout,` `TimeUnit unit)` 如果在给定的等待时间内，此信号量有可用的许可并且当前线程未被中断，则从此信号量获取一个许可。

Semaphore实现的功能就类似厕所有5个坑，假如有10个人要上厕所，那么同时只能有多少个人去上厕所呢？同时只能有5个人能够占用，当5个人中的任何一个人让开后，其中等待的另外5个人中又有一个人可以占用了。另外等待的5个人中可以是随机获得优先机会，也可以是按照先来后到的顺序获得机会，这取决于构造Semaphore对象时传入的参数选项。

三、Semaphore应用场景

实现互斥锁功能：（对应上面的单值）单个信号量的Semaphore对象可以实现互斥锁的功能，并且可以是由一个线程获得了“锁”，再由另一个线程释放“锁”，这可应用于死锁恢复的一些场合。

控制同时访问的个数：（对应上面的多值）Semaphore维护了当前访问的个数，提供同步机制，控制同时访问的个数。在数据结构中链表可以保存“无限”的节点，用Semaphore可以实现有限大小的链表。另外重入锁 ReentrantLock 也可以实现该功能，但实现上要复杂些。

示例-1：Semaphore可以做到一个deadlock recovery的示例

package com.dxz.semaphore2;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
class WorkThread1 extends Thread {
    private Semaphore semaphore1, semaphore2;
    public WorkThread1(Semaphore semaphore1, Semaphore semaphore2) {
        this.semaphore1 = semaphore1;
        this.semaphore2 = semaphore2;
    }
    public void releaseSemaphore2() {
        System.out.println(Thread.currentThread().getId() + " 释放Semaphore2");
        semaphore2.release();
    }
    public void run() {
        try {
            semaphore1.acquire(); // 先获取Semaphore1
            System.out.println(Thread.currentThread().getId() + " 获得Semaphore1");
            TimeUnit.SECONDS.sleep(5); // 等待5秒让WorkThread1先获得Semaphore2
            semaphore2.acquire();// 获取Semaphore2
            System.out.println(Thread.currentThread().getId() + " 获得Semaphore2");
            System.out.println(Thread.currentThread().getId() + "线程开始干活");
            System.out.println(Thread.currentThread().getId() + "线程事情干完");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println(Thread.currentThread().getId() + "释放semaphore1、semaphore2");
            semaphore1.release();
            semaphore2.release();
        }
    }
}
class WorkThread2 extends Thread {
    private Semaphore semaphore1, semaphore2;
    public WorkThread2(Semaphore semaphore1, Semaphore semaphore2) {
        this.semaphore1 = semaphore1;
        this.semaphore2 = semaphore2;
    }
    public void run() {
        try {
            semaphore2.acquire();// 先获取Semaphore2
            System.out.println(Thread.currentThread().getId() + " 获得Semaphore2");
            TimeUnit.SECONDS.sleep(5);// 等待5秒，让WorkThread1先获得Semaphore1
            semaphore1.acquire();// 获取Semaphore1
            System.out.println(Thread.currentThread().getId() + " 获得Semaphore1");
            System.out.println(Thread.currentThread().getId() + "线程开始干活");
            System.out.println(Thread.currentThread().getId() + "线程事情干完");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println(Thread.currentThread().getId() + "释放semaphore2、semaphore1");
            semaphore2.release();
            semaphore1.release();
        }
    }
}
public class SemphoreTest {
    public static void main(String[] args) throws InterruptedException {
        Semaphore semaphore1 = new Semaphore(1);
        Semaphore semaphore2 = new Semaphore(1);
        new WorkThread1(semaphore1, semaphore2).start();
        new WorkThread2(semaphore1, semaphore2).start();
        System.out.println("2个线程已经启动");
        // 此时已经陷入了死锁，WorkThread1持有semaphore1的许可，请求semaphore2的许可
        // WorkThread2持有semaphore2的许可，请求semaphore1的许可
        TimeUnit.SECONDS.sleep(10);
        //在主线程释放semaphore1或semaphore2,解决死锁
        System.out.println("10秒后，主线程释放semaphore1");
        semaphore1.release();
    }
}

输出：

10 获得Semaphore1

2个线程已经启动

11 获得Semaphore2

10秒后，主线程释放semaphore1

2个线程运行完毕

11 获得Semaphore1

11线程开始干活

11线程事情干完

11释放semaphore2、semaphore1

10 获得Semaphore2

10线程开始干活

10线程事情干完

10释放semaphore1、semaphore2

这即符合文档中说的，通过一个非owner的线程来实现死锁恢复，但如果你使用的是Lock则做不到，可以把代码中的两个信号量换成两个锁对象试试。很明显，前面也验证过了，要使用Lock.unlock()来释放锁，首先你得拥有这个锁对象，因此非owner线程（事先没有拥有锁）是无法去释放别的线程的锁对象。

示例2-Semaphore限流：

Semaphore 可以用来限流（流量控制），在一些公共资源有限的场景下，Semaphore 可以派上用场。比如在做日志清洗时，可能有几十个线程在并发清洗，但是将清洗的数据存入到数据库时，可能只给数据库分配了 10 个连接池，这样两边的线程数就不对等了，我们必须保证同时只能有 10 个线程获取数据库链接，否则就会存在大量线程无法链接上数据库。

用 Semaphore 信号量来模拟这操作，代码如下：

package com.dxz.semaphore1;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
/**
 * semaphore 信号量，可以限流 模拟并发数据库操作，同时有三十个请求，但是系统每秒只能处理 5 个
 */
public class SemaphoreDemo {
    private static final int THREAD_COUNT = 30;
    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
    // 初始化信号量，个数为 5
    private static Semaphore s = new Semaphore(5);
    public static void main(String[] args) {
        for (int i = 0; i < THREAD_COUNT; i++) {
            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 获取许可
                        s.acquire();
                        System.out.println(Thread.currentThread().getName() + " 完成数据库操作 ,"
                                + new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date()));
                        System.out.println("线程" + Thread.currentThread().getName() + "进入，当前已有"
                                + (5 - s.availablePermits()) + "个并发");
                        // 休眠两秒钟，效果更直观
                        Thread.sleep(2000);
                        // 释放许可
                        s.release();
                    } catch (InterruptedException e) {
                    }
                }
            });
        }
        // 关闭连接池
        threadPool.shutdown();
    }
}

运行效果：

线程pool-1-thread-28进入，当前已有5个并发

pool-1-thread-11 完成数据库操作 ,2021-04-15 02:02:15

线程pool-1-thread-11进入，当前已有5个并发

pool-1-thread-15 完成数据库操作 ,2021-04-15 02:02:17

线程pool-1-thread-15进入，当前已有5个并发

pool-1-thread-19 完成数据库操作 ,2021-04-15 02:02:17

线程pool-1-thread-19进入，当前已有5个并发

pool-1-thread-23 完成数据库操作 ,2021-04-15 02:02:17

线程pool-1-thread-23进入，当前已有4个并发

pool-1-thread-27 完成数据库操作 ,2021-04-15 02:02:17

线程pool-1-thread-27进入，当前已有4个并发

pool-1-thread-10 完成数据库操作 ,2021-04-15 02:02:17

线程pool-1-thread-10进入，当前已有5个并发

pool-1-thread-14 完成数据库操作 ,2021-04-15 02:02:19

线程pool-1-thread-14进入，当前已有5个并发

pool-1-thread-18 完成数据库操作 ,2021-04-15 02:02:19

线程pool-1-thread-18进入，当前已有5个并发

从结果中，可以看出，每秒只有 5 个线程在执行，这符合我们的预期。

限流2：下面的Demo中申明了一个只有5个许可的Semaphore，而有20个线程要访问这个资源，通过acquire()和release()获取和释放访问许可。

package com.dxz.semaphore;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class TestSemaphore {
    public static void main(String[] args) {
        // 线程池
        ExecutorService exec = Executors.newCachedThreadPool();
        // 只能5个线程同时访问
        final Semaphore semp = new Semaphore(5);
        // 模拟20个客户端访问
        for (int index = 0; index < 20; index++) {
            final int NO = index;
            Runnable run = new Runnable() {
                public void run() {
                    try {
                        // 获取许可
                        semp.acquire();
                        System.out.println("Accessing: " + NO);
                        Thread.sleep((long) (Math.random() * 1000));
                        // 访问完后，释放
                        semp.release();
                        System.out.println("semp.availablePermits()==" + semp.availablePermits());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            exec.execute(run);
        }
        // 退出线程池
        exec.shutdown();
    }
}

执行结果如下：

Accessing: 0
Accessing: 2
Accessing: 4
Accessing: 6
Accessing: 8
semp.availablePermits()==0
Accessing: 10
Accessing: 12
semp.availablePermits()==0
semp.availablePermits()==1
Accessing: 14
semp.availablePermits()==1
Accessing: 16
semp.availablePermits()==1
Accessing: 18
semp.availablePermits()==1
Accessing: 1
semp.availablePermits()==1
Accessing: 3
semp.availablePermits()==1
Accessing: 5
semp.availablePermits()==1
Accessing: 7
semp.availablePermits()==1
Accessing: 9
semp.availablePermits()==1
Accessing: 11
Accessing: 13
semp.availablePermits()==0
semp.availablePermits()==1
Accessing: 15
semp.availablePermits()==1
Accessing: 17
semp.availablePermits()==1
Accessing: 19
semp.availablePermits()==1
semp.availablePermits()==2
semp.availablePermits()==3
semp.availablePermits()==4
semp.availablePermits()==5