并发工具类（三）控制并发线程的数量 Semphore

时间 2019-11-08

标签并发工具控制线程数量 semphore 繁體版

原文原文链接

前言

JDK中为了处理线程之间的同步问题，除了提供锁机制以外，还提供了几个很是有用的并发工具类：CountDownLatch、CyclicBarrier、Semphore、Exchanger、Phaser；
CountDownLatch、CyclicBarrier、Semphore、Phaser 这四个工具类提供一种并发流程的控制手段；而Exchanger工具类则提供了在线程之间交换数据的一种手段。html

简介

Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它经过协调各个线程，以保证合理的使用公共资源。

不少年以来，我都以为从字面上很难理解Semaphore所表达的含义，只能把它比做是控制流量的红绿灯，好比XX马路要限制流量，只容许同时有一百辆车在这条路上行使，其余的都必须在路口等待，因此前一百辆车会看到绿灯，能够开进这条马路，后面的车会看到红灯，不能驶入XX马路，可是若是前一百辆中有五辆车已经离开了XX马路，那么后面就容许有5辆车驶入马路，这个例子里说的车就是线程，驶入马路就表示线程在执行，离开马路就表示线程执行完成，看见红灯就表示线程被阻塞，不能执行。

应用场景

Semaphore能够用于作java

流量控制

，特别公用资源有限的应用场景，好比数据库链接。假若有一个需求，要读取几万个文件的数据，由于都是IO密集型任务，咱们能够启动几十个线程并发的读取，可是若是读到内存后，还须要存储到数据库中，而数据库的链接数只有10个，这时咱们必须控制只有十个线程同时获取数据库链接保存数据，不然会报错没法获取数据库链接。这个时候，咱们就可使用Semaphore来作流控，代码以下：

public class SemaphoreTest {

    private static final int THREAD_COUNT = 30;

    private static ExecutorService threadPool = Executors
            .newFixedThreadPool(THREAD_COUNT);

    private static Semaphore s = new Semaphore(10);

    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("save data");
                        s.release();
                    } catch (InterruptedException e) {
                    }
                }
            });
        }

        threadPool.shutdown();
    }
}
复制代码

在代码中，虽然有30个线程在执行，可是只容许10个并发的执行。Semaphore的构造方法Semaphore(int permits) 接受一个整型的数字，表示可用的许可证数量。Semaphore(10)表示容许10个线程获取许可证，也就是最大并发数是10。Semaphore的用法也很简单，首先线程使用Semaphore的acquire()获取一个许可证，使用完以后调用release()归还许可证。还能够用tryAcquire()方法尝试获取许可证。
spring

Semphore的方法摘要

一、获取许可

API中提供了多种的方式获取锁：数据库

能够获取一个、多个许可；
提供阻塞、非阻塞、超时的方式获取许可；
除了可中断、还提供一个非中断的方式获取锁；

public void acquire() throws InterruptedException
今后信号量获取一个许可，在提供一个许可前一直将线程阻塞，不然线程被bash

中断。

public void acquire(int permits) throws InterruptedException
数据结构

获取多个许可。

今后信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞，或者线程已被并发

中断。

public void acquireUninterruptibly()
今后信号量中获取许可，在有可用的许可前将其阻塞。ide

不可中断。

public void acquireUninterruptibly(int permits)
工具

获取多个许可。

今后信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞。测试

不可中断。

public boolean tryAcquire()
仅在调用时此信号量存在一个可用许可，才从信号量获取许可。

非阻塞的方式尝试获取许可。

public boolean tryAcquire(int permits)
仅在调用时此信号量中有给定数目的许可时，才今后信号量中获取这些许可。
public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException
若是在给定的等待时间内此信号量有可用的全部许可，而且当前线程未被中断，则今后信号量获取给定数目的许可。

超时等待获取许可

public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException
若是在给定的等待时间内，此信号量有可用的许可而且当前线程未被中断，则今后信号量获取一个许可。

二、许可的释放

public void release( )：
释放一个许可，将其返回给信号量。
public void release(int permits)
释放给定数目的许可，将其返回到信号量。

三、提供的监控方法

public int availablePermits( )
返回此信号量中当前可用的许可数
public int drainPermits()
获取并返回当即可用的全部许可
public final int getQueueLength()
返回正在等待获取的线程的估计数目。该值仅是估计的数字，由于在此方法遍历内部数据结构的同时，线程的数目可能动态地变化。此方法用于监视系统状态，不用于同步控制。
public final boolean hasQueuedThreads()
查询是否有线程正在等待获取。
public boolean isFair()
若是此信号量的公平设置为 true，则返回 true。

protected 方法：

protected Collection
返回一个 collection，包含可能等待获取的线程。由于在构造此结果的同时实际的线程 set 可能动态地变化，因此返回的 collection 仅是尽力的估计值。所返回 collection 中的元素没有特定的顺序。
protected void reducePermits(int reduction)
根据指定的缩减量减少可用许可的数目。此方法在使用信号量来跟踪那些变为不可用资源的子类中颇有用

@ Example 获取、释放多个许可

try {
    Semaphore semaphore = new Semaphore(5);
    //获取一个许可
    semaphore.acquire();
    //一次性获取4个许可
    semaphore.acquire(4);
    System.out.println("Semaphore 剩下的许可数量："+semaphore.availablePermits());
    //一次性释放5个许可
    semaphore.release(5);
    System.out.println("Semaphore 剩下的许可数量："+semaphore.availablePermits());
    //再释放5个许可
    semaphore.release();
    semaphore.release();
    semaphore.release(3);
    System.out.println("Semaphore 剩下的许可数量："+semaphore.availablePermits());
   
    } catch (InterruptedException e) {
    e.printStackTrace();
复制代码

运行结果：

Semaphore 剩下的许可数量：0
Semaphore 剩下的许可数量：5
Semaphore 剩下的许可数量：10
复制代码

从上面的运行结果能够看出，

构造方法的 new Semaphore(5)中参数5并非最终的许可数量，能够经过release（）方法增长许可数量。

本人测试用例

package com.wxx.demo;

import com.wxx.demo.util.IdUtiles;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;

@RunWith(SpringRunner.class)
//@SpringBootTest(classes = LeisureWebApplication.class)
public class TaskTest {

    @Test
    public void taskTest(){

        Runnable task = new Runnable() {

            int count = 0;

            @Override
            public void run() {

                count ++;
                try{
                    String id = IdUtiles.creatId();
                    System.out.println(id);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
                System.out.println(count);
                System.out.println("Thread : " + Thread.currentThread().getId());

                try {
                    Thread.sleep(1000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }

            }
        };

        double executeTime = this.executeTime(100, task);
        System.out.println("执行时间： " + executeTime);
    }

    private double executeTime(int taskCount,Runnable task){

        CountDownLatch start = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(taskCount);

        for (int i = 0; i < taskCount ; i++) {
            Thread thread = new Thread() {

                public void run(){
                    try {
                        start.await();

                        try {
                            task.run();
                        }finally {
                            end.countDown();
                        }

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

            thread.start();
        }

        long startTime = System.nanoTime();

        //开启开关
        start.countDown();

        long endTime = System.nanoTime();

        return endTime - startTime;
    }
}
复制代码

package com.wxx.demo.util;

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

/**
 * @Author : leisure
 * @Date : 2019/1/17
 */
public class IdUtiles {

    private static String lead = "leisure";
    private static int Guid = 100;
    private static Semaphore semaphore = new Semaphore(5,false);
    /**
     * 建立以字符串打头结尾自增的惟一id
     * @return
     */
    public static synchronized String creatId() throws InterruptedException{
        //测试控制方法内的并发线程数 测试放开synchronized
        //semaphore.acquire();
        semaphore.tryAcquire(1,1000, TimeUnit.MILLISECONDS);
        int i = semaphore.availablePermits();
        System.out.println("当前可用许可" + i);


        int i1 = semaphore.drainPermits();
        System.out.println("当前当即可用的许可" + i1);

        boolean b = semaphore.hasQueuedThreads();
        System.out.println("当前是否有线程等待" + b);

        boolean fair = semaphore.isFair();
        System.out.println("当前信号是否公平" + fair);
        long l = System.currentTimeMillis();

        int queueLength = semaphore.getQueueLength();
        System.out.println("等待线程数" + queueLength);
        Thread.sleep(100);
        Guid += 1;

        String format = new SimpleDateFormat("yyyy").format(l);

        if (Guid > 999){
            Guid = 100;
        }

        String id = lead + format + l + Guid;

        semaphore.release();
        return id;
    }
}
复制代码

注：文章源地址：https://www.cnblogs.com/jinggod/p/8494246.html