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

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

估计也没多少人看个人博客,哈哈,那么我仍是会记下来,天空未曾留下个人痕迹,但我已飞过,而在博客园留下了个人痕迹~编程

 一、Semaphore的初步使用多线程

  Semaphore是什么,能作什么?并发

    Semaphore 是 synchronized 的增强版,做用是控制线程的并发数量。就这一点而言,单纯的synchronized 关键字是实现不了的。框架

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

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() 也能够运行,可是不是并发执行了 
        }
    }
}

运行结果:ide

 

 

 实践证实,确实是同一个时刻只有一个线程能访问,那若是把 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学习

 二、方法 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 结果:

 

 

 

三、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 中
        }
    }

四、其余一些常有工具方法

  availablePermits()  方法在前面用过,表示返回 Semaphore 对象中的当前可用许可数,此方法一般用于调试,由于许可数量(通路)多是实时在改变的。

  drainPermits() 方法可获取并返回当即可用的全部许可(通路)个数,并将可用许可置为0。

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

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

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

  这里就不写代码校验了,大家能够在 SemaphoreService 或者 SemaphoreService2 中加入这个信息试一下。

五、线程公平性

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

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

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

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

六、方法 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 运行结果:

 

 

 

七、多进路-多处理 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 实例中,就至关于没有锁

相关文章
相关标签/搜索