【java并发核心一】Semaphore 的使用思路

最近在看一本书《Java并发编程 核心方法与框架》，打算一边学习一边把学习的经验记下来，所粘贴的代码都是我运行过的，大家一起学习，欢迎吐槽。

估计也没多少人看我的博客，哈哈，那么我还是会记下来，天空不曾留下我的痕迹，但我已飞过，而在博客园留下了我的痕迹~

1、Semaphore的初步使用

　　Semaphore是什么，能做什么？

　　　　Semaphore 是 synchronized 的加强版，作用是控制线程的并发数量。就这一点而言，单纯的synchronized 关键字是实现不了的。

　　直接看例子吧，这个例子包含3个类，一个是线程类，一个是 Semaphore 关键代码类，一个类是主main方法类：

package com.cd.concurrent.semaphore;

public class MyThread extends Thread {
    private SemaphoreService service;

    public MyThread(String name, SemaphoreService service) {
        super();
        this.setName(name);
        this.service = service;
    }

    @Override
    public void run() {
        this.service.doSomething();
    }
}
package com.cd.concurrent.semaphore;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Semaphore;

public class SemaphoreService {

    private static SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    private Semaphore semaphore = new Semaphore(1);// 同步关键类，构造方法传入的数字是多少，则同一个时刻，只运行多少个进程同时运行制定代码

    public void doSomething() {
        try {
            /**
             * 在 semaphore.acquire() 和 semaphore.release()之间的代码，同一时刻只允许制定个数的线程进入，
             * 因为semaphore的构造方法是1，则同一时刻只允许一个线程进入，其他线程只能等待。
             * */
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static String getFormatTimeStr() {
        return sf.format(new Date());
    }
}
package com.cd.concurrent.semaphore;

public class SemaphoreTest {
    public static void main(String args[]) {
        SemaphoreService service = new SemaphoreService();
        for (int i = 0; i < 10; i++) {
            MyThread t = new MyThread("thread" + (i + 1), service);
            t.start();// 这里使用 t.run() 也可以运行，但是不是并发执行了
        }
    }
}

运行结果：

实践证明，确实是同一个时刻只有一个线程能访问，那如果把 Semaphore 的构造方法入参改成 2 呢，修改 SemaphoreService.java 文件：

package com.cd.concurrent.semaphore;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Semaphore;

public class SemaphoreService {

    private static SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    private Semaphore semaphore = new Semaphore(2);// 同步关键类，构造方法传入的数字是多少，则同一个时刻，只运行多少个进程同时运行制定代码

    public void doSomething() {
        try {
            /**
             * 在 semaphore.acquire() 和 semaphore.release()之间的代码，同一时刻只允许制定个数的线程进入，
             * 因为semaphore的构造方法是2，则同一时刻只允许2个线程进入，其他线程等待。
             * */
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static String getFormatTimeStr() {
        return sf.format(new Date());
    }
}

运行SemaphoreTest，结果如下：

验证OK

2、方法 acquire( int permits ) 参数作用，及动态添加 permits 许可数量　　

　　acquire( int permits ) 中的参数是什么意思呢？可以这么理解， new Semaphore(6) 表示初始化了 6个通路， semaphore.acquire(2) 表示每次线程进入将会占用2个通路，semaphore.release(2) 运行时表示归还2个通路。没有通路，则线程就无法进入代码块。

　　而上面的代码中，semaphore.acquire() +  semaphore.release()  在运行的时候，其实和 semaphore.acquire(1) + semaphore.release(1)  效果是一样的。　　

　　上代码：

　　还是3个代码，线程类没有变，用的是上面的线程类，重新写了另外两个类：

package com.cd.concurrent.semaphore;

import java.util.concurrent.Semaphore;

public class SemaphoreService2 extends SemaphoreService { // 之所以继承 SemaphoreService，仅仅是为了使用父类的打印时间的方法 0.0

    private Semaphore semaphore = new Semaphore(6);// 6表示总共有6个通路

    public void doSomething() {
        try {
            semaphore.acquire(2); // 2 表示进入此代码，就会消耗2个通路，2个通路从6个中扣除
            System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
            semaphore.release(2); // 释放占用的 2 个通路
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public int availablePermits() {    // 查看可用通路数
        return semaphore.availablePermits();
    }
}
package com.cd.concurrent.semaphore;

public class SemaphoreTest2 {
    public static void main(String args[]) {
        SemaphoreService2 service = new SemaphoreService2(); // 使用总 6 通路，每个线程占用2通路
        for (int i = 0; i < 10; i++) {
            MyThread t = new MyThread("thread" + (i + 1), service);
            t.start();// 这里使用 t.run() 也可以运行，但是不是并发执行了
            System.out.println("可用通路数：" + service.availablePermits());
        }
    }
}

运行结果：

如果 acquire 的数量大于 release 的数量，则 通路迟早会被使用完，如果线程比较多，得不到后续运行，出现线程堆积内存，最终java进程崩掉；如果 acquire 的数量小于 release 的数量，就会出现并发执行的线程越来越多（换句话说，处理越来越快），最终也有可能出现问题。

　　比如，象上面的代码，SemaphoreService2.java 中 semaphore.release(2) 如果改成 semaphore.release(1) 则 就会出现有5个线程得不到运行堆积的情况，可以算一下：6-2-2-2+1+1+1=3，运行完一个回合后，还剩3个通路，3-2+1，第二回合，还剩2个通路，2-2+1=1，第3个回合，还剩一个通路，不足以运行任何一个线程。

　　把上面说的用代码实现一下，修改 SemaphoreService2.java 如下：

package com.cd.concurrent.semaphore;

import java.util.concurrent.Semaphore;

public class SemaphoreService2 extends SemaphoreService { // 之所以继承 SemaphoreService，仅仅是为了使用父类的打印时间的方法 0.0

    private Semaphore semaphore = new Semaphore(6);// 6表示总共有6个通路

    public void doSomething() {
        try {
            semaphore.acquire(2); // 2 表示进入此代码，就会消耗2个通路，2个通路从6个中扣除
            System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
            semaphore.release(1); // 释放占用的 1 个通路
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public int availablePermits() {
        return semaphore.availablePermits();
    }
}

运行 SemaphoreTest2 结果：

3、acquire 的不可中断实现

　　仔细看一下上面的代码，semaphore.acquire() 和 semaphore.acquire(int permits) 是会抛出异常 InterruptedException 的，如果在 acquire 和 release 之间的代码是一个比较慢和复制的运算，如内存占用过多，或者栈深度很深等，jvm会中断这块代码。

　　如何才能不让 jvm 中断 代码执行呢？

　　答案是：使用 acquireUninterruptibly() 替换acquire()、使用 acquireUninterruptibly(int permits) 替换 acquire(int permits) 。

　　acquireUninterruptibly 不会抛出 InterruptedException ，一个代码块一时执行不完，还会继续等待执行。

　　个人觉得，不要随便使用 acquireUninterruptibly ，因为 jvm 中断执行，是自身的一种自我保护机制，保证 java 进程的正常，除了特殊情况必须用 acquireUninterruptibly 外，都应该 使用 acquire ，同时，改进一下 SemaphoreService2 的 doSomething 方法，将 release 放到 finally 块 中，如下。　　

public void doSomething() {
        try {
            semaphore.acquire(2); // 2 表示进入此代码，就会消耗2个通路，2个通路从6个中扣除
            System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release(2); // release 放到 finally 中
        }
    }

4、其他一些常有工具方法

　　availablePermits()  方法在前面用过，表示返回 Semaphore 对象中的当前可用许可数，此方法通常用于调试，因为许可数量（通路）可能是实时在改变的。

　　drainPermits() 方法可获取并返回立即可用的所有许可（通路）个数，并将可用许可置为0。

　　getQueueLength() 获取等待许可的线程个数。

　　hasQueuedThreads() 判断有没有线程在等待这个许可。

　　getQueueLength() 和 hasQueuedThreads() 都是在判断当前有没有等待许可的线程信息时使用。

　　这里就不写代码校验了，你们可以在 SemaphoreService 或者 SemaphoreService2 中加入这个信息试一下。

5、线程公平性

　　上面用的 Semaphore  构造方法是 Semaphore semaphore = new Semaphore(int permits)

　　其实，还有一个构造方法： Semaphore semaphore = new Semaphore(int permits , boolean isFair)

　　isFair 的意思就是，是否公平，获得锁的顺序与线程启动顺序有关，就是公平，先启动的线程，先获得锁。isFair 不能100% 保证公平，只能是大概率公平。

　　isFair 为 true，则表示公平，先启动的线程先获得锁。

6、方法 tryAcquire() 、 tryAcquire(int permits)、 tryAcquire(int permits , long timeout , TimeUint unit) 的使用：

　　tryAcquire 方法，是 acquire 的扩展版，tryAcquire 作用是尝试得获取通路，如果未传参数，就是尝试获取一个通路，如果传了参数，就是尝试获取 permits 个 通路 、在指定时间 timeout  内 尝试 获取 permits 个通路。

　　上代码试试看：

　　3个类，线程类未变，以下是修改了的两个类：

package com.cd.concurrent.semaphore;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class SemaphoreService3 extends SemaphoreService { // 之所以继承 SemaphoreService，仅仅是为了使用父类的打印时间的方法 0.0

    private Semaphore semaphore = new Semaphore(6, true);// 6表示总共有6个通路，true 表示公平

    public void doSomething() {
        try {
            if (semaphore.tryAcquire(2, 3, TimeUnit.SECONDS)) { // 在 3秒 内 尝试获取 2 个通路

                System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr()
                        + "，当前是否有进程等待：" + semaphore.hasQueuedThreads() + "，等待进程数：" + semaphore.getQueueLength());
                semaphore.release(2); // 释放占用的 2 个通路
            } else {
                System.out.println(Thread.currentThread().getName() + ":doSomething 没有获取到锁-准备退出-" + getFormatTimeStr());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public int availablePermits() {
        return semaphore.availablePermits();
    }
}
package com.cd.concurrent.semaphore;

public class SemaphoreTest3 {
    public static void main(String args[]) {
        SemaphoreService3 service = new SemaphoreService3(); // 使用总 6 通路，每个线程占用2通路，尝试获取锁
        for (int i = 0; i < 10; i++) {
            MyThread t = new MyThread("thread" + (i + 1), service);
            t.start();
        }
    }
}

SemaphoreTest3 运行结果：

7、多进路-多处理 vs 多进路-单处理

　　在上面的代码中，我们之所以可以实现单处理，是因为在上面的所有线程都共有了同一个 Semaphore 来进行进程处理，那么如果 Semaphore 本身就是进程的一部分呢，会怎么样呢？

　　比如，修改 第一个例子中的 SemaphoreTest  如下：

package com.cd.concurrent.semaphore;

public class SemaphoreTest {
    public static void main(String args[]) {
        for (int i = 0; i < 10; i++) {
            SemaphoreService service = new SemaphoreService();
            MyThread t = new MyThread("thread" + (i + 1), service);
            t.start();// 这里使用 t.run() 也可以运行，但是不是并发执行了
        }
    }
}

运行 SemaphoreTest 结果：

所有线程同时执行了。

　　如果 SemaphoreTest  类不进行修改，如何实现第一个例子 中的 单处理呢？

　　也简单，修改 SemaphoreService ，代码如下：

package com.cd.concurrent.semaphore;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Semaphore;

public class SemaphoreService {

    private static SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    private Semaphore semaphore = new Semaphore(2);// 同步关键类，构造方法传入的数字是多少，则同一个时刻，只运行多少个进程同时运行制定代码

    public void doSomething() {
        try {
            /**
             * 在 semaphore.acquire() 和 semaphore.release()之间的代码，同一时刻只允许制定个数的线程进入，
             * 因为semaphore的构造方法是1，则同一时刻只允许一个线程进入，其他线程只能等待。
             * */
            semaphore.acquire();

            doSomethingMain(); // 将主要处理部分封装成一个方法

            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static synchronized void doSomethingMain() throws InterruptedException {
        System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
        Thread.sleep(2000);
        System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
    }

    public static String getFormatTimeStr() {
        return sf.format(new Date());
    }
}
注意：doSomethingMain() 方法必须是 static synchronized 的才行，因为 多线程调用的话，static 方法是类方法，这样 synchronized 同步 才能针对整个类同步，否则 就只能针对单线程多个地方调用同步。

　　修改 SemaphoreService ，运行 SemaphoreTest 结果：

　运行达到想要的效果。

　　这里，抛出一个问题，上面的代码，不用 synchronized 实现，而使用 ReentrantLock 来实现，按理说会更好的，原因如下：

　　　　synchronized 是 jvm 层面的实现，ReentrantLock 是 jdk 层面的实现，synchronized 的缺点如下：

　　　　1）不能响应中断；

　　　　2）同一时刻不管是读还是写都只能有一个线程对共享资源操作，其他线程只能等待

　　　　3）锁的释放由虚拟机来完成，不用人工干预，不过此即使缺点也是优点，优点是不用担心会造成死锁，缺点是由可能获取到锁的线程阻塞之后其他线程会一直等待，性能不高。

　　而lock接口的提出就是为了完善synchronized的不完美的，首先lock是基于jdk层面实现的接口，和虚拟机层面不是一个概念；其次对于lock对象中的多个方法的调用，可以灵活控制对共享资源变量的操作，不管是读操作还是写操作

　　那么上面的代码如果使用 ReentrantLock 来实现，岂不是更好吗？好，修改 SemaphoreService：

package com.cd.concurrent.semaphore;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.ReentrantLock;

public class SemaphoreService {

    private static SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    private Semaphore semaphore = new Semaphore(2);// 同步关键类，构造方法传入的数字是多少，则同一个时刻，只运行多少个进程同时运行制定代码

    private ReentrantLock lock = new ReentrantLock();

    public void doSomething() {
        try {
            /**
             * 在 semaphore.acquire() 和 semaphore.release()之间的代码，同一时刻只允许制定个数的线程进入，
             * 因为semaphore的构造方法是1，则同一时刻只允许一个线程进入，其他线程只能等待。
             * */
            semaphore.acquire();

            lock.lock();
            doSomethingMain(); // 将主要处理部分封装成一个方法            

            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    private void doSomethingMain() throws InterruptedException {
        System.out.println(Thread.currentThread().getName() + ":doSomething start-" + getFormatTimeStr());
        Thread.sleep(2000);
        System.out.println(Thread.currentThread().getName() + ":doSomething end-" + getFormatTimeStr());
    }

    public static String getFormatTimeStr() {
        return sf.format(new Date());
    }
}

运行 SemaphoreTest 结果：

和预期的不一样呀，10个线程基本是同时执行了，那么问题出在哪里呢？

因为使用的不是同一个 SemaphoreService 对象实例，所有是多个锁分别加在了多个 SemaphoreService 实例中，就相当于没有锁

原文地址：https://www.cnblogs.com/zhaoyan001/p/11447955.html

时间： 2024-10-11 00:04:10

【java并发核心一】Semaphore 的使用思路的相关文章

Java 并发核心机制

目录   一.J.U.C 简介  二.synchronized  三.volatile  四.CAS  五.ThreadLocal  参考资料 ?? 本文以及示例源码已归档在 javacore 一.J.U.C 简介 Java 的 java.util.concurrent 包(简称 J.U.C)中提供了大量并发工具类,是 Java 并发能力的主要体现(注意,不是全部,有部分并发能力的支持在其他包中).从功能上,大致可以分为: 原子类 - 如:AtomicInteger.AtomicIntegerA

java并发编程之Semaphore

信号量(Semaphore).有时被称为信号灯.是在多线程环境下使用的一种设施, 它负责协调各个线程, 以保证它们可以正确.合理的使用公共资源. 一个计数信号量.从概念上讲,信号量维护了一个许可集.如有必要.在许可可用前会堵塞每个 acquire(),然后再获取该许可.每个 release() 加入一个许可.从而可能释放一个正在堵塞的获取者. 可是.不使用实际的许可对象,Semaphore 仅仅对可用许可的号码进行计数,并採取对应的行动.拿到信号量的线程能够进入代码.否则就等待.通过acquir

【Java并发核心九】并发集合框架

1.List接口:ArrayList 和 Vector ArrayList不是线程安全的,Vector是线程安全的,Vector有一个子类,可实现后进先出(LIFO)的对象堆栈(LinkedList 也是List接口的实现类). 2.Set接口:HashSet 和 TreeSet Set接口最常见的实现类是HashSet,HashSet默认是以无序的方式组织元素的,而LinkedHashSet可以有序组织元素: Treeset不仅实现了Set接口,还实现了SortedSet和NavigableS

【Java并发核心四】Executor 与 ThreadPoolExecutor

Executor 和 ThreadPoolExecutor 实现的是线程池,主要作用是支持高并发的访问处理. Executor 是一个接口,与线程池有关的大部分类都实现了此接口. ExecutorService 是 Executor 的子接口:AbstractExecutorService 是 ExecutorService 的实现类,但是是抽象类. ThreadPoolExecutor 是 AbstractExecutorService 的子类,可实例化. Executors 是一个工厂类,用

java中的信号量Semaphore

       Semaphore当前在多线程环境下被扩放使用,操作系统的信号量是个很重要的概念,在进程控制方面都有应用.Java 并发库 的Semaphore 可以很轻松完成信号量控制,Semaphore可以控制某个资源可被同时访问的个数,通过 acquire() 获取一个许可,如果没有就等待,而 release() 释放一个许可.比如在Windows下可以设置共享文件的最大客户端访问个数. Semaphore实现的功能就类似厕所有5个坑,假如有10个人要上厕所,那么同时只能有多少个人去上厕所呢

Java并发编程小总结：CountDownLatch、CyclicBarrier和Semaphore

Java并发编程小总结:CountDownLatch.CyclicBarrier和Semaphore这几个类都是在JUC下,也就是java.util.concurrent包下.这两天学习了一下并发编程中的三个类的使用和一些应用场景,所以做一下记录和总结,方便自己日后再查看复现. 1.CountDownLatch.这个类的核心思想总结为8个字“秦灭6国,一统华夏”.它可以实现的是一个类似计数器的功能,与CyclicBarrier的思想正好相反.是一个减法操作.CountDownLatch有且只有一

【Java并发编程实战】—–“J.U.C”：Semaphore

信号量Semaphore是一个控制访问多个共享资源的计数器,它本质上是一个"共享锁". Java并发提供了两种加锁模式:共享锁和独占锁.前面LZ介绍的ReentrantLock就是独占锁.对于独占锁而言,它每次只能有一个线程持有,而共享锁则不同,它允许多个线程并行持有锁,并发访问共享资源. 独占锁它所采用的是一种悲观的加锁策略,  对于写而言为了避免冲突独占是必须的,但是对于读就没有必要了,因为它不会影响数据的一致性.如果某个只读线程获取独占锁,则其他读线程都只能等待了,这种情况下就限

Java 并发编程：核心理论

并发编程是Java程序员最重要的技能之一,也是最难掌握的一种技能.它要求编程者对计算机最底层的运作原理有深刻的理解,同时要求编程者逻辑清晰.思维缜密,这样才能写出高效.安全.可靠的多线程并发程序.本系列会从线程间协调的方式(wait.notify.notifyAll).Synchronized及Volatile的本质入手,详细解释JDK为我们提供的每种并发工具和底层实现机制.在此基础上,我们会进一步分析java.util.concurrent包的工具类,包括其使用方式.实现源码及其背后的原理.本

Java并发工具类（三）控制并发线程数的Semaphore

作用 Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源. 简介 Semaphore也是一个线程同步的辅助类,可以维护当前访问自身的线程个数,并提供了同步机制.使用Semaphore可以控制同时访问资源的线程个数,例如,实现一个文件允许的并发访问数. 主要方法摘要: void acquire():从此信号量获取一个许可,在提供一个许可前翼子将线程阻塞,否则线程被中断. void release():释放一个许可,将其返回给信号量. in