首页 文章 精选 留言 我的

精选列表

搜索[源码学习],共10010篇文章
优秀的个人博客,低调大师

java源码 - CountDownLatch

开篇 CountDownLatch是一个同步工具类,用来协调多个线程之间的同步,或者说起到线程之间的通信(而不是用作互斥的作用)。 CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后,再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后,计数器的值就会减一。当计数器的值为0时,表示所有的线程都已经完成了任务,然后在CountDownLatch上等待的线程就可以恢复执行任务。 CountDownLatch是一次性的,计数器的值只能在构造方法中初始化一次,之后没有任何机制再次对其设置值,当CountDownLatch使用完毕后,它不能再次被使用。 CountDownLatch的用法 CountDownLatch典型用法1:某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为n new CountDownLatch(n) ,每当一个任务线程执行完毕,就将计数器减1 countdownlatch.countDown(),当计数器的值变为0时,在CountDownLatch上 await() 的线程就会被唤醒。一个典型应用场景就是启动一个服务时,主线程需要等待多个组件加载完毕,之后再继续执行。 CountDownLatch典型用法2:实现多个线程开始执行任务的最大并行性。注意是并行性,不是并发,强调的是多个线程在某一时刻同时开始执行。类似于赛跑,将多个线程放到起点,等待发令枪响,然后同时开跑。做法是初始化一个共享的CountDownLatch(1),将其计数器初始化为1,多个线程在开始执行任务前首先 coundownlatch.await(),当主线程调用 countDown() 时,计数器变为0,多个线程同时被唤醒。 CountDownLatch的demo public class CountDownLatchDemo { public static void main(String[] args) throws InterruptedException{ CountDownLatch countDownLatch = new CountDownLatch(2){ @Override public void await() throws InterruptedException { super.await(); System.out.println(Thread.currentThread().getName() + " count down is ok"); } }; Thread thread1 = new Thread(new Runnable() { @Override public void run() { //do something try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " is done"); countDownLatch.countDown(); } }, "thread1"); Thread thread2 = new Thread(new Runnable() { @Override public void run() { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " is done"); countDownLatch.countDown(); } }, "thread2"); thread1.start(); thread2.start(); countDownLatch.await(); } CountDownLatch的类定义 CountDownLatch内部包含Sync类。 CountDownLatch内部包含Sync类的对象sync。 Sync类继承自AQS(神奇的AQS),构造函数设置AQS的state值为等待值。 public class CountDownLatch { private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } } private final Sync sync; public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } } CountDownLatch的等待过程 CountDownLatch通过await()进入等待。 CountDownLatch通过await(long timeout, TimeUnit unit)进入超时等待。 public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } CountDownLatch的await()过程 await()通过sync.acquireSharedInterruptibly()获锁。 acquireSharedInterruptibly通过tryAcquireShared()尝试获锁。 tryAcquireShared()判断获锁成功与否的依据是AQS的state的值是否为零。 获锁失败后通过doAcquireSharedInterruptibly()进入锁等待队列CLH。 public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 尝试获锁失败 if (tryAcquireShared(arg) < 0) // doAcquireSharedInterruptibly(arg); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } CountDownLatch的await(long timeout, TimeUnit unit)过程 await(long timeout, TimeUnit unit)通过sync.tryAcquireSharedNanos()获锁。 tryAcquireSharedNanos()通过doAcquireSharedNanos()尝试获锁。 tryAcquireShared()判断获锁成功与否的依据是AQS的state的值是否为零。 获锁失败后通过doAcquireSharedNanos()进入锁等待队列CLH,和doAcquireSharedInterruptibly()方法相比增加了超时检测机制,通过LockSupport.parkNanos()实现超时。 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout); } private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (nanosTimeout <= 0L) return false; final long deadline = System.nanoTime() + nanosTimeout; final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return true; } } nanosTimeout = deadline - System.nanoTime(); if (nanosTimeout <= 0L) return false; if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } CountDownLatch的唤醒过程 CountDownLatch通过sync.releaseShared(1)释放锁实现state的递减 tryReleaseShared()方法判断锁状态state==0,递减后值为0说明锁已经被释放。 releaseShared()释放锁成功后通过doReleaseShared()方法唤醒所有等待线程。 doReleaseShared()唤醒锁的过程是一个传播性的唤醒,通过线程A唤醒线程B,然后由线程B唤醒线程C的传播性依次唤醒所有等待线程。 public void countDown() { sync.releaseShared(1); } public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } protected boolean tryReleaseShared(int releases) { for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } } 总结 CountDownLatch的工作原理,总结起来就两点(基于AQS实现): 初始化锁状态的值为需要等待的线程数。 判断锁状态是否已经释放,如果锁未释放所有等待锁的线程就会进入等待的CLH队列。 如果锁状态已经释放,那么就会通过传播性唤醒所有的等待线程。

优秀的个人博客,低调大师

java源码 - CyclicBarrier

开篇 CyclicBarrier是一个同步工具类,它允许一组线程互相等待,直到到达某个公共屏障点。与CountDownLatch不同的是该barrier在释放等待线程后可以重用,所以称它为循环(Cyclic)的屏障(Barrier)。 CyclicBarrier支持一个可选的Runnable命令,在一组线程中的最后一个线程到达之后(但在释放所有线程之前),该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态,此屏障操作很有用。 CyclicBarrier的内部实现逻辑基于ReentrantLock实现,可以理解为ReentrantLock的上层应用者,通过ReentrantLock的Condtion实现线程的休眠和唤醒。 CyclicBarrier用法demo public class Test { public static void main(String[] args) { int N = 4; CyclicBarrier barrier = new CyclicBarrier(N); for(int i=0;i<N;i++) new Writer(barrier).start(); } static class Writer extends Thread{ private CyclicBarrier cyclicBarrier; public Writer(CyclicBarrier cyclicBarrier) { this.cyclicBarrier = cyclicBarrier; } @Override public void run() { System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据..."); try { Thread.sleep(5000); //以睡眠来模拟写入数据操作 System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕,等待其他线程写入完毕"); cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); }catch(BrokenBarrierException e){ e.printStackTrace(); } System.out.println("所有线程写入完毕,继续处理其他任务..."); } } } 线程Thread-0正在写入数据... 线程Thread-3正在写入数据... 线程Thread-2正在写入数据... 线程Thread-1正在写入数据... 线程Thread-2写入数据完毕,等待其他线程写入完毕 线程Thread-0写入数据完毕,等待其他线程写入完毕 线程Thread-3写入数据完毕,等待其他线程写入完毕 线程Thread-1写入数据完毕,等待其他线程写入完毕 所有线程写入完毕,继续处理其他任务... 所有线程写入完毕,继续处理其他任务... 所有线程写入完毕,继续处理其他任务... 所有线程写入完毕,继续处理其他任务... CyclicBarrier类定义 parties记录一共等待执行个数,count记录依然等待执行的个数。 barrierCommand记录所有待执行的完成后由最后一个线程执行的完成的命令。 Generation的代的概念来实现CyclicBarrier的复用。 构造函数负责初始化parties、count、barrierCommand的核心变量。 public class CyclicBarrier { // 代的类定义 private static class Generation { boolean broken = false; } // 内部通过ReentrantLock实现线程安全的等待 private final ReentrantLock lock = new ReentrantLock(); // 内部通过Lock的condition实现所有waiter的信号通知 private final Condition trip = lock.newCondition(); // 所有等待执行的个数 private final int parties; // 所有等待线程都完成任务后由最后一个线程执行的命令 private final Runnable barrierCommand; // 通过代的概念实现复用 private Generation generation = new Generation(); // 还在等待的个数 private int count; // 核心构造函数 public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; } public CyclicBarrier(int parties) { this(parties, null); } CyclicBarrier工作原理 CyclicBarrier通过ReentrantLock来保证线程休眠和唤醒的通信。 在执行过程中会对等待计数进行减一操作,值不为0当前线程进入休眠等待其他线程唤醒 在执行过程中会对等待计数进行减一操作,值为0当前线程直接执行barrierCommand并且通过nextGeneration方法唤醒其他等待线程 线程的休眠和唤醒都是基于ReentrantLock来实现的 public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { // 通过lock来保证线程安全 final ReentrantLock lock = this.lock; lock.lock(); try { final Generation g = generation; // 判断generation过期的情况 if (g.broken) throw new BrokenBarrierException(); // 判断线程中断情况 if (Thread.interrupted()) { breakBarrier(); throw new InterruptedException(); } // 递减待执行的个数计数 int index = --count; // 所有待执行任务完成后执行barrierCommand命令 if (index == 0) { // tripped boolean ranAction = false; try { // barrierCommand命令不为null的时候执行该命令 final Runnable command = barrierCommand; if (command != null) command.run(); // 已经执行了barrierCommand ranAction = true; // 重置generation用以复用并且唤醒所有等待的线程 // private void nextGeneration() { // trip.signalAll(); // count = parties; // generation = new Generation(); // } nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // 如果count的值不为0,那么当前线程就开始进入等待 // 外层通过lock占用锁,内层通过wait()进入休眠并释放锁 for (;;) { try { if (!timed) // private final Condition trip = lock.newCondition(); trip.await(); else if (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { if (g == generation && ! g.broken) { breakBarrier(); throw ie; } else { // We're about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. Thread.currentThread().interrupt(); } } // 各种后置处理逻辑 if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } } 参考文章 Java并发编程:CountDownLatch、CyclicBarrier和Semaphore

优秀的个人博客,低调大师

fishhook源码分析

最早了解到fishhook是看了下面两篇文章之后,顿时让我觉得这是一个非常好的东西。总共210行代码,收获了1500+个star,神作啊。 iOS Lazy Binding,使用fishhook拦截NSSetUncaughtExceptionHandler函数解决NSUncaughtExceptionHandler被修改的问题。 聊聊苹果的Bug - iOS 10 nano_free Crash,通过fishhook替换malloc相关的函数尝试解决crash。 OC的runtime非常强大,可以玩很多黑魔法。而操作系统也会提供一些基础设施,比如Solaris/Mac DTrace和Linux systemtap,通过编写脚本分析各种系统调用的情况。相比之下,C语言没有runtime,也玩不出什么花来。不过fishhook提供了一种很好

优秀的个人博客,低调大师

java源码-AtomicInteger

开篇 AtomicInteger位于java.util.concurrent.atomic包下,是java提供给的可以保证数据的原子性操作的一个类。 Atomicxxxx系列主要核心在于Unsafe这个类的运用保证线程安全,而Unsafe这个类应该是通过JNI调用的底层实现。 关于unsafe类可以看看揭秘sun.misc.Unsafe,虽然我还是没怎么看懂。记住unsafe这个东西很重要,不过据说在jdk9之后应该会被弃用了。 AtomicInteger类构造器 AtomicInteger类构造器有两个: 无参构造函数采用默认值初始化为0 有参数构造函数直接用initialValue来value的 AtomicInteger的关键逻辑在于static代码快中通过unsafe接口初始化value的内存地址,后续直接通过内存地址进行操作。 另外我们需要注意到value是用volatile进行修饰保证变量的可见性,这个有空一定要仔细研究研究。 public class AtomicInteger extends Number implements java.io.Serializable { private static final long serialVersionUID = 6214790243416807050L; // setup to use Unsafe.compareAndSwapInt for updates private static final Unsafe unsafe = Unsafe.getUnsafe(); private static final long valueOffset; static { try { valueOffset = unsafe.objectFieldOffset (AtomicInteger.class.getDeclaredField("value")); } catch (Exception ex) { throw new Error(ex); } } private volatile int value; // value值初始化为initialValue public AtomicInteger(int initialValue) { value = initialValue; } // value的值为默认值为0 public AtomicInteger() { } } AtomicInteger的get操作 AtomicInteger的get操作还是比较有意思的,总共非几大类: get()方法直接返回值 getAndIncrement()、getAndDecrement()、getAndAdd()、getAndUpdate()先返回旧值后对旧值执行加减操作。 incrementAndGet()、decrementAndGet()、addAndGet()、updateAndGet()先执行加减操作后返回新值。 底层操作unsafe.getAndAddInt()的实现细节借助于unsafe的compareAndSwapInt()方法保证只有在旧值为v的情况下才能更新为v+delta值。 public final int getAndAddInt(Object o, long offset, int delta) { int v; do { v = getIntVolatile(o, offset); } while (!compareAndSwapInt(o, offset, v, v + delta)); return v; } --------------------------------------------------------------- public final int get() { return value; } public final int getAndIncrement() { return unsafe.getAndAddInt(this, valueOffset, 1); } public final int getAndDecrement() { return unsafe.getAndAddInt(this, valueOffset, -1); } public final int getAndAdd(int delta) { return unsafe.getAndAddInt(this, valueOffset, delta); } public final int getAndUpdate(IntUnaryOperator updateFunction) { int prev, next; do { prev = get(); next = updateFunction.applyAsInt(prev); } while (!compareAndSet(prev, next)); return prev; } public final int getAndAccumulate(int x, IntBinaryOperator accumulatorFunction) { int prev, next; do { prev = get(); next = accumulatorFunction.applyAsInt(prev, x); } while (!compareAndSet(prev, next)); return prev; } --------------------------------------------------------------- public final int incrementAndGet() { return unsafe.getAndAddInt(this, valueOffset, 1) + 1; } public final int decrementAndGet() { return unsafe.getAndAddInt(this, valueOffset, -1) - 1; } public final int addAndGet(int delta) { return unsafe.getAndAddInt(this, valueOffset, delta) + delta; } public final int updateAndGet(IntUnaryOperator updateFunction) { int prev, next; do { prev = get(); next = updateFunction.applyAsInt(prev); } while (!compareAndSet(prev, next)); return next; } public final int accumulateAndGet(int x, IntBinaryOperator accumulatorFunction) { int prev, next; do { prev = get(); next = accumulatorFunction.applyAsInt(prev, x); } while (!compareAndSet(prev, next)); return next; } AtomicInteger的set操作 AtomicInteger的set操作基本上也分为两大类: 直接set设置value值 通过unsafe的api接口实现原子性写操作 public final int getAndAddInt(Object o, long offset, int delta) { int v; do { v = getIntVolatile(o, offset); } while (!compareAndSwapInt(o, offset, v, v + delta)); return v; } ------------------------------------------------------------------ public final void set(int newValue) { value = newValue; } public final void lazySet(int newValue) { unsafe.putOrderedInt(this, valueOffset, newValue); } public final int getAndSet(int newValue) { return unsafe.getAndSetInt(this, valueOffset, newValue); } public final boolean compareAndSet(int expect, int update) { return unsafe.compareAndSwapInt(this, valueOffset, expect, update); } public final boolean weakCompareAndSet(int expect, int update) { return unsafe.compareAndSwapInt(this, valueOffset, expect, update); } 参考文章 揭秘sun.misc.Unsafe非阻塞同步算法与CAS(Compare and Swap)无锁算法为什么volatile不能保证原子性而Atomic可以?彻底了解内存屏障

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

用户登录
用户注册