首页 > > 程序设计 > Java技术 >

AQS 框架之 LockSupport 线程阻塞工具类

2018-06-18 03:31:59来源：未知 阅读 ()

新老客户大回馈,云服务器低至5折

■ 前言

　　并发包一直是 JDK 里面比较难理解的，同时也是很精美的语言，膜拜下 Doug Li 大神。作者不敢长篇大论，只求循序渐进地把并发包通过理论和实战 (代码) 的方式介绍给大家。

其实做每一件事都是挺难的，不过只要下笔就不会瞻前顾后。谢谢大家的鼓励帮助，感谢我的好基友KIRA~  好的，热身先从 LockSupport 开始吧~

 

■ LockSupport 综述

  • 定义： LockSupport 是一个线程阻塞工具类，可用于在线程内任意位置让线程阻塞和释放
  • 作用： LockSupport 通常不会被直接使用，更多是作为锁实现的基础工具类
  • 实现： LockSupport 底层依赖UnSafe实现，即 park() 和 unpark() 原语方法，通过"许可"替代状态
  • 使用： park方法用于线程等待"许可"，unpark方法用于为线程提供"许可"
  • 补充1：由于"许可"的存在，当出现一个线程调用park方法，其他线程调用unpark方法时，会保持活跃
  • 补充2：若开JVM篇的话笔者会从JVM源码角度再次解析park和unpark的底层实现，其实质用mutex和condition维护一个_counter(park->0，unpark->1)的变量，即"许可"是一次性的
  • 补充3：此番为 AQS 框架之综述 (赶制中) 的子番

 

■ LockSupport 数据结构

　1. 类定义

public class LockSupport

　

   2. 构造器

//私有构造器，不能被实例化 -- 实质就是个工作类，只能调用静态方法
private LockSupport() {} // Cannot be instantiated.

 

  3. UnSafe

// Hotspot implementation via intrinsics API
private static final sun.misc.Unsafe UNSAFE;
//用于记录线程被谁阻塞的，用于线程监控和分析工具来定位原因，其表示parkBlocker在内存的偏移量
//之所以用偏移量是因为parkBlockerOffset被赋值时线程必须是阻塞的，阻塞时直接调方法无效只能走内存
private static final long parkBlockerOffset;
private static final long SEED;
private static final long PROBE;
private static final long SECONDARY;
static {
    try {
        UNSAFE = sun.misc.Unsafe.getUnsafe();
        Class<?> tk = Thread.class;
        //获取指定变量的内存偏移量
        parkBlockerOffset = UNSAFE.objectFieldOffset
            (tk.getDeclaredField("parkBlocker"));
        SEED = UNSAFE.objectFieldOffset
            (tk.getDeclaredField("threadLocalRandomSeed"));
        PROBE = UNSAFE.objectFieldOffset
            (tk.getDeclaredField("threadLocalRandomProbe"));
        SECONDARY = UNSAFE.objectFieldOffset
            (tk.getDeclaredField("threadLocalRandomSecondarySeed"));
    } catch (Exception ex) { throw new Error(ex); }
}

 ■ Unsafe 综述

  • 作用： Unsafe是个后门类，封装了一些类似指针的操作，提供了一些可以直接操控内存和线程的底层操作
  • 使用： Unsafe被JDK广泛用于nio包和并发包中，但是不建议在生产环境使用，风险太大
  • 不安全： 不安全指的是指针的操作不安全(Java因此才把指针去掉)，若指针指错位置或计算指针偏移量出错，结果不可想象，比如说覆盖了别人的内存，那可能就GG思密达了...
  • 补充1： 有机会开JVM番的话，笔者会从JVM源码角度重新解析Unsafe一些重要方法的实现
  • 补充2： 此番为 AQS 框架之综述 (赶制中) 的子番

 

■ Unsafe 数据结构

　1. 类定义

public final class Unsafe

   2. 构造器

//私有构造器 --单例模式
private Unsafe() {}

  3. 重要变量

private static final Unsafe theUnsafe;
public static final int INVALID_FIELD_OFFSET = -1;
public static final int ARRAY_BOOLEAN_BASE_OFFSET;
public static final int ARRAY_BYTE_BASE_OFFSET;
public static final int ARRAY_SHORT_BASE_OFFSET;
public static final int ARRAY_CHAR_BASE_OFFSET;
public static final int ARRAY_INT_BASE_OFFSET;
public static final int ARRAY_LONG_BASE_OFFSET;
public static final int ARRAY_FLOAT_BASE_OFFSET;
public static final int ARRAY_DOUBLE_BASE_OFFSET;
public static final int ARRAY_OBJECT_BASE_OFFSET;
public static final int ARRAY_BOOLEAN_INDEX_SCALE;
public static final int ARRAY_BYTE_INDEX_SCALE;
public static final int ARRAY_SHORT_INDEX_SCALE;
public static final int ARRAY_CHAR_INDEX_SCALE;
public static final int ARRAY_INT_INDEX_SCALE;
public static final int ARRAY_LONG_INDEX_SCALE;
public static final int ARRAY_FLOAT_INDEX_SCALE;
public static final int ARRAY_DOUBLE_INDEX_SCALE;
public static final int ARRAY_OBJECT_INDEX_SCALE;
public static final int ADDRESS_SIZE;
private static native void registerNatives();
static {
    registerNatives();
    Reflection.registerMethodsToFilter(Unsafe.class, new String[]{"getUnsafe"});
    theUnsafe = new Unsafe();//单例模式 -饿汉式
    ARRAY_BOOLEAN_BASE_OFFSET = theUnsafe.arrayBaseOffset(boolean[].class);
    ARRAY_BYTE_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class);
    ARRAY_SHORT_BASE_OFFSET = theUnsafe.arrayBaseOffset(short[].class);
    ARRAY_CHAR_BASE_OFFSET = theUnsafe.arrayBaseOffset(char[].class);
    ARRAY_INT_BASE_OFFSET = theUnsafe.arrayBaseOffset(int[].class);
    ARRAY_LONG_BASE_OFFSET = theUnsafe.arrayBaseOffset(long[].class);
    ARRAY_FLOAT_BASE_OFFSET = theUnsafe.arrayBaseOffset(float[].class);
    ARRAY_DOUBLE_BASE_OFFSET = theUnsafe.arrayBaseOffset(double[].class);
    ARRAY_OBJECT_BASE_OFFSET = theUnsafe.arrayBaseOffset(Object[].class);
    ARRAY_BOOLEAN_INDEX_SCALE = theUnsafe.arrayIndexScale(boolean[].class);
    ARRAY_BYTE_INDEX_SCALE = theUnsafe.arrayIndexScale(byte[].class);
    ARRAY_SHORT_INDEX_SCALE = theUnsafe.arrayIndexScale(short[].class);
    ARRAY_CHAR_INDEX_SCALE = theUnsafe.arrayIndexScale(char[].class);
    ARRAY_INT_INDEX_SCALE = theUnsafe.arrayIndexScale(int[].class);
    ARRAY_LONG_INDEX_SCALE = theUnsafe.arrayIndexScale(long[].class);
    ARRAY_FLOAT_INDEX_SCALE = theUnsafe.arrayIndexScale(float[].class);
    ARRAY_DOUBLE_INDEX_SCALE = theUnsafe.arrayIndexScale(double[].class);
    ARRAY_OBJECT_INDEX_SCALE = theUnsafe.arrayIndexScale(Object[].class);
    ADDRESS_SIZE = theUnsafe.addressSize();
}

  4. 重要方法

//获得给定对象内存偏移量的int值
public native int getInt(Object var1, long var2);
//设置给定对象内存偏移量的int值
public native void putInt(Object var1, long var2, int var4);
public native Object getObject(Object var1, long var2);
public native void putObject(Object var1, long var2, Object var4);
//....还有Boolean、Byte、Char、Short、Long、Float、Double的get\set....
//内存分配、释放
//分配内存
public native long allocateMemory(long var1);
//扩充内存
public native long reallocateMemory(long var1, long var3);
public native void setMemory(Object var1, long var2, long var4, byte var6);
public void setMemory(long var1, long var3, byte var5) {
    this.setMemory((Object)null, var1, var3, var5);
}
//拷贝内存
public native void copyMemory(Object var1, long var2, Object var4, long var5, long var7);
public void copyMemory(long var1, long var3, long var5) {
    this.copyMemory((Object)null, var1, (Object)null, var3, var5);
}
//释放内存
public native void freeMemory(long var1);
//获取字段在对象中的内存偏移量
public native long staticFieldOffset(Field var1);
public native long objectFieldOffset(Field var1);
public native Object staticFieldBase(Field var1);
public native void ensureClassInitialized(Class<?> var1);
//数组元素定位
//arrayBaseOffset 和 arrayIndexScale 搭配使用可以定位数组中每个元素在内存中的位置
//获取数组第一个元素的偏移地址
public native int arrayBaseOffset(Class<?> var1);
//获取数组的转换因子，也就是数组中元素的增量地址
public native int arrayIndexScale(Class<?> var1);
public native int addressSize();
public native int pageSize();
//类定义
public native Class<?> defineClass(String var1, byte[] var2, int var3, int var4, ClassLoader var5, ProtectionDomain var6);
public native Class<?> defineClass(String var1, byte[] var2, int var3, int var4);
public native Class<?> defineAnonymousClass(Class<?> var1, byte[] var2, Object[] var3);
//创建实例
public native Object allocateInstance(Class<?> var1) throws InstantiationException;
//Synchronized同步块的指令实现 1.8版的全部是@Deprecated
public native void monitorEnter(Object var1);
public native void monitorExit(Object var1);
public native boolean tryMonitorEnter(Object var1);
//异常抛出
public native void throwException(Throwable var1);
//CAS操作
/**
* 比较obj的offset处内存位置中的值和期望的值，如果相同则更新，此更新是不可中断的
* @param obj 需要更新的对象
* @param offset obj中整型field的偏移量
* @param expect 希望field中存在的值
* @param update 如果期望值expect与field的当前值相同，设置filed的值为这个新值
* @return 如果field的值被更改返回true
*/
public final native boolean compareAndSwapObject(Object obj, long offset, Object expect, Object update);
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);
//获取给定对象的指定类型值，支持volatile load语义
public native Object getObjectVolatile(Object var1, long var2);
public native void putObjectVolatile(Object var1, long var2, Object var4);
public native int getIntVolatile(Object var1, long var2);
//设置给定对象的int值，支持volatile load语义
public native void putIntVolatile(Object var1, long var2, int var4);
//....还有Boolean、Byte、Char、Short、Long、Float、Double的volatile级别的get\put....
public native void putDoubleVolatile(Object var1, long var2, double var4);
public native void putOrderedObject(Object var1, long var2, Object var4);
public native void putOrderedInt(Object var1, long var2, int var4);
public native void putOrderedLong(Object var1, long var2, long var4);
//LockSupport类的原语支持-挂起和唤醒某个线程
public native void unpark(Object var1);
public native void park(boolean var1, long var2);
public native int getLoadAverage(double[] var1, int var2);
//提供线程安全的add和set操作
public final int getAndAddInt(Object var1, long var2, int var4) {
int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
    return var5;
}
public final int getAndSetInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var4));
    return var5;
}
//...还有Long和Object的线程安全的add和set操作...
//栅栏支持
public native void loadFence();
public native void storeFence();
public native void fullFence();

  5. 禁用的工厂方法

@CallerSensitive
public static Unsafe getUnsafe() {
    Class var0 = Reflection.getCallerClass();
    //该方法用于判断调用者的类加载器是否是系统核心加载器（即Bootstrap加载器）
    if(!VM.isSystemDomainLoader(var0.getClassLoader())) {
        throw new SecurityException("Unsafe");
    } else {
        return theUnsafe;
    }
}
  • 我们先显性调用该工厂方法查看一下调用结果
package concurrent;
import sun.misc.Unsafe;
public class UnsafeDemo {
    public static void main(String[] args) {
        Unsafe.getUnsafe();
    }
}
-------------------
//输出：
Exception in thread "main" java.lang.SecurityException: Unsafe
    at sun.misc.Unsafe.getUnsafe(Unsafe.java:90)
    at concurrent.UnsafeDemo.main(UnsafeDemo.java:7)
//分析：可以发现直接调用的话会直接抛出安全异常，原因是类加载器是AppClassLoader而并非是BootstrapLoader

 

  • 根据Java 类加载器的工作原理，应用程序的类由AppLoader加载，而系统核心类由BootstrapLoader加载
  • 当一个类的类加载器为null时，说明它是由BootstrapLoader加载的，即此类是系统核心类(比如rt.jar包中的类)
  • 当一个类无法被BootstrapLoader加载时，其类加载器通常为AppClassLoader，即是属于自定义类

 

■ Unsafe 反射获取

/** 
  * 我们可以通过反射机制获取Unsafe 的一个实例
  */
public static Unsafe getUnsafe(){
    try {
        //通过反射获取Unsafe的theUnsafe变量，即Unsafe实例对象
        Field f = Unsafe.class.getDeclaredField("theUnsafe");
        f.setAccessible(true);
        //注意field是static属性
        //参见：private static final Unsafe theUnsafe;
        return (Unsafe) f.get(null);
    } catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}

 

  4. permit (许可)

  • LockSupport 和每个使用它的线程都与一个permit关联，某种意义上可认为是 Semaphore 类，但区别于Semaphores，permit至多只有一个，并不能被累加（即重复调动unpark也不会累加，最多为1）
  • permit 相当于一个开关(只有0和1两个值)，默认为0，执行过程如下：
    • 调用unpark方法，permit+1，即permit=1
    • 调用park方法，permit被消费-1，即permit=0，同时park方法理解返回
    • 再次调用park方法，线程会被阻塞(此时permit=0，线程无许可可用，直到permit=1之前都会被阻塞)

 

■ LockSupport 数据结构

　1. setBlocker / getBlocker

 1 /**
 2  * Returns the blocker object supplied to the most recent
 3  * invocation of a park method that has not yet unblocked, or null
 4  * if not blocked.  The value returned is just a momentary
 5  * snapshot -- the thread may have since unblocked or blocked on a
 6  * different blocker object.
 7  *  返回提供给最近一次尚未解除阻塞的被park方法调用的blocker对象，如果该调用未阻塞，则返回null
 8  * @param t the thread
 9  * @return the blocker
10  * @throws NullPointerException if argument is null
11  * @since 1.6
12  */
13 public static Object getBlocker(Thread t) {
14     if (t == null)
15         throw new NullPointerException();
16     return UNSAFE.getObjectVolatile(t, parkBlockerOffset);
17 }
18 /**
19   * This object is recorded while the thread is blocked to permit monitoring and diagnostic 
20   * tools to identify the reasons that threads are blocked.
21   * 此对象在线程受阻塞时被记录，以允许监视工具和诊断工具确定线程受阻塞的原因
22   */
23 private static void setBlocker(Thread t, Object arg) {
24     // Even though volatile, hotspot doesn't need a write barrier here.
25     UNSAFE.putObject(t, parkBlockerOffset, arg);
26 }
  • 通过线程 Dump 查看一下，可以很明显的看到阻塞的堆栈信息，但其实信息是差不多一样的

 

· 2. park

  • 作用：该方法用于等待"许可"，调用时可能发生以下两种情况：
    • 当"许可"可用时，立即返回并且消费这个许可(将许可变成不可用)
    • 当"许可"不可用时，当前线程可能被阻塞 java.lang.Thread.State : WAITING parking
  • 使用： 由于park方法可能在任何时候"无理由"返回，因此通常会在循环中使用(在返回之前再次检查条件)
  • 适用： park方法是"busy wait"(忙碌等待)的一种优化 (即不需要在自旋上浪费太多时间)，但它必须与 unpark 配对使用才更高效
  • 注意： park方法的许可默认是被占用的，在unpark之前调用会获取不到许可而被阻塞
public static void park() {
    UNSAFE.park(false, 0L);
}
//纳秒级超时返回
public static void parkNanos(long nanos) {
    if (nanos > 0)
        UNSAFE.park(false, nanos);
}
//毫秒级限时等待
//注意这里的时间需要使用系统时间加上需要等待的时间
//LockSupport.parkUntil(System.currentTimeMillis() + 3000);
public static void parkUntil(long deadline) {
    UNSAFE.park(true, deadline);
}
//三种形式的 park 还各自支持一个 blocker 对象参数
//建议最好使用这些形式，而不是不带此参数的原始形式
//在锁实现中提供的作为 blocker 的普通参数是 this
public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(false, 0L);
    setBlocker(t, null);
}
public static void parkNanos(Object blocker, long nanos) {
    if (nanos > 0) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, nanos);
        setBlocker(t, null);
    }
}
public static void parkUntil(Object blocker, long deadline) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(true, deadline);
    setBlocker(t, null);
}

 

· 3. unpark

  • 作用： 该方法用于提供"许可"，会将还不可用的"许可"变成可用
  • 注意： 由于 park方法默认是许可占有并阻塞线程，因此调用 park之前最好先调用 unpark (当然因为 park\unpark 的顺序解耦性，所以前后执行顺序无所谓，只是代码上最好遵循 先释放再获取 的规则)
/**
  * 注意：必须指定一个线程(但无所谓该线程是否park)，将尝试释放其可能拥有的许可
  */
public static void unpark(Thread thread) {
    if (thread != null)
        UNSAFE.unpark(thread);
}

 

· 4. LockSupport 不可重入

  • 不可重入： LockSupport 不可重入，当一个线程多次调用 park 方法，线程将被第二个 park 方法阻塞
 1 public static void main(String[] args) {
 2     LockSupport.unpark(Thread.currentThread());//我们直接用主线程
 3     System.out.println("执行unpark");
 4     LockSupport.park();
 5     System.out.println("执行第一次park");
 6     LockSupport.park();
 7     System.out.println("执行第二次park");
 8     while (true);
 9 }
10 ---------------------
11 //输出：
12 执行unpark
13 执行第一次park
14 //分析：通过打印结果可以发现第7行其实并没有被打印，根据下面的图片可以看到线程在第7行(对应图片的第16行)上阻塞
  • 使用 jstack 命令查看一下线程状态，会发现线程是 WAITING 状态，即等待阻塞，而且还是被 park方法阻塞

 

 ·5. LockSupport 与中断

  • 中断响应： LockSupport支持中断响应，线程调用park阻塞时仍能够响应中断请求，但不会抛出InterruptedException异常
public static void main(String[] args) {
    Thread thread = new Thread(() -> {
        long start = System.currentTimeMillis();
        while ((System.currentTimeMillis() - start) <= 1000);//空转1s
        System.out.println("空转1s结束");
        LockSupport.park();//等待"许可"
        System.out.println(Thread.currentThread().getName() + "是否被中断：" 
            + Thread.currentThread().isInterrupted());
    },"kira");
    thread.start();
    thread.interrupt();//中断线程
}
---------------------
//输出：
空转1s结束
kira是否被中断：true
//分析：通过先中断线程再park可以发现可获取中断响应，同时并没有抛出任何异常

 

 ■ suspend()  VS  wait()  VS  park()

　 1. suspend() VS wait()

  • suspend() 不会释放锁，wait()会释放锁同时还支持超时处理

    2. suspend() VS wait()

  • LockSupport 解决了suspend()不释放锁从而容易死锁的问题，比如resume()方法被阻塞时，即其他线程在调用 resume()方法之前获取同步锁时被阻塞而导致 resume()方法无法执行进而导致死锁

    3. park() VS wait()

  • LockSupport 不需要先获得某个对象的锁，也不会排除 InterruptedException异常
  • unpark 方法可以先于park方法调用，其没有方法调用的时序问题
  • wait/notify 机制有个问题在于线程调用notify方法去唤醒其他线程时，需要保证需被唤醒线程必须被wait方法阻塞，否则被唤醒线程会永远处于 WAITING 状态，同时notify方法只能唤醒一个线程，当同时有多个线程在同一个对象上 wait 等待，就只能有一个线程可以被唤醒(不能指定)
  • park/unpark 机制通过引入单个"许可"的概念实现对线程同步的解耦，线程间无须关心对方的状态，因为不需要一个变量专门用于存储状态

 

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有

上一篇：Hibernate从入门到上手（纯java project、Maven版本hibernate）

下一篇：JavaWeb项目架构之Redis分布式日志队列

相关文章

IDC资讯： 主机资讯 注册资讯 托管资讯 vps资讯 网站建设

网站运营： 建站经验 策划盈利 搜索优化 网站推广 免费资源

网站联盟： 联盟新闻 联盟介绍 联盟点评 网赚技巧

行业资讯： 搜索引擎 网络游戏 电子商务 广告传媒

网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它

服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护

软件技巧： 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷

网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash

程序设计： Java技术 C/C++ VB delphi

网络知识： 网络协议 网络安全 网络管理 组网方案 Cisco技术

操作系统： Win2000 WinXP Win2003 Mac OS Linux FreeBSD

热门词条
最新资讯
热门关注
热门标签