时间：2023-06-20 06:28:38   /   来源： 博客园      /   点击数：()

【资料图】

JUC同步锁原理源码解析六----ExchangerExchangerExchanger的来源

A synchronization point at which threads can pair and swap elements within pairs.  Each thread presents some object on entry to the {@link #exchange exchange} method, matches with a partner thread, and receives its partner"s object on return. An Exchanger may be viewed as a bidirectional form of a {@link SynchronousQueue}.

​JDK中对Exchanger的定义是在一个同步线程点，配对的线程可以交换彼此的属性数据。每一个线程提交对象数据并调用exchange方法，匹配到一个线程并接受该线程携带的数据返回。Exchanger可以被当成一个双向的同步队列。当然Exchanger并不是说只有两个线程进行匹配，它也可以进行多对多的匹配，但是只有成对的线程可以匹配并交换数据成功。

Exchanger的底层实现

​Exchanger的底层实现依旧依赖于CAS的自旋锁操作，通过cas保证原子性的操作

2.Exchanger基本使用

public class ExchangerDemo {    static Exchanger exchanger = new Exchanger<>();    public static void main(String[] args) {        new Thread(()->{            String s = "T1";            try {                s = exchanger.exchange(s);            } catch (InterruptedException e) {                e.printStackTrace();            }            System.out.println(Thread.currentThread().getName() + " " + s);        }, "t1").start();        new Thread(()->{            String s = "T2";            try {                s = exchanger.exchange(s);            } catch (InterruptedException e) {                e.printStackTrace();            }            System.out.println(Thread.currentThread().getName() + " " + s);        }, "t2").start();    }}

Exchanger类

public class Exchanger {    /**     * The byte distance (as a shift value) between any two used slots     * in the arena.  1 << ASHIFT should be at least cacheline size.     */    private static final int ASHIFT = 7;    /**     * The maximum supported arena index. The maximum allocatable     * arena size is MMASK + 1. Must be a power of two minus one, less     * than (1<<(31-ASHIFT)). The cap of 255 (0xff) more than suffices     * for the expected scaling limits of the main algorithms.     */    private static final int MMASK = 0xff;    /**     * Unit for sequence/version bits of bound field. Each successful     * change to the bound also adds SEQ.     */    private static final int SEQ = MMASK + 1;    /** The number of CPUs, for sizing and spin control */    private static final int NCPU = Runtime.getRuntime().availableProcessors();    /**     * The maximum slot index of the arena: The number of slots that     * can in principle hold all threads without contention, or at     * most the maximum indexable value.     */    static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1;    /**     * The bound for spins while waiting for a match. The actual     * number of iterations will on average be about twice this value     * due to randomization. Note: Spinning is disabled when NCPU==1.     */    private static final int SPINS = 1 << 10;    /**     * Value representing null arguments/returns from public     * methods. Needed because the API originally didn"t disallow null     * arguments, which it should have.     */    private static final Object NULL_ITEM = new Object();    /**     * Sentinel value returned by internal exchange methods upon     * timeout, to avoid need for separate timed versions of these     * methods.     */    private static final Object TIMED_OUT = new Object();    /** The corresponding thread local class */    static final class Participant extends ThreadLocal {        public Node initialValue() { return new Node(); }    }    /**     * Per-thread state     */    private final Participant participant;    /**     * Elimination array; null until enabled (within slotExchange).     * Element accesses use emulation of volatile gets and CAS.     */    private volatile Node[] arena;    /**     * Slot used until contention detected.     */    private volatile Node slot;    /**     * The index of the largest valid arena position, OR"ed with SEQ     * number in high bits, incremented on each update.  The initial     * update from 0 to SEQ is used to ensure that the arena array is     * constructed only once.     */    private volatile int bound;}

Node类

@sun.misc.Contended static final class Node {//@sun.misc.Contended 进行缓存行填充，防止数据移植刷新缓存行，造成性能损耗    int index;              // Arena index    int bound;              // Last recorded value of Exchanger.bound    int collides;           // Number of CAS failures at current bound    int hash;               // Pseudo-random for spins    Object item;            // This thread"s current item    volatile Object match;  // Item provided by releasing thread    volatile Thread parked; // Set to this thread when parked, else null}

Exchanger的构造器

public Exchanger() {    participant = new Participant();}

exchange()方法

public V exchange(V x) throws InterruptedException {    Object v;    Object item = (x == null) ? NULL_ITEM : x; // 将对象赋值给item    if ((arena != null ||//表示有多个线程在竞争匹配         (v = slotExchange(item, false, 0L)) == null) &&//slot匹配的对象返回空，slot表示的一对一的匹配。        ((Thread.interrupted() || // 线程是否发生中断          (v = arenaExchange(item, false, 0L)) == null)))//arena表示发生多线程竞争，如果匹配失败        throw new InterruptedException();//抛出中断异常    return (v == NULL_ITEM) ? null : (V)v;//返回匹配后并交换的数据}

exchange(V x, long timeout, TimeUnit unit)方法:

public V exchange(V x, long timeout, TimeUnit unit)    throws InterruptedException, TimeoutException {    Object v;    Object item = (x == null) ? NULL_ITEM : x;// 将对象赋值给item    long ns = unit.toNanos(timeout);//超时的时间换算成纳秒    if ((arena != null ||         (v = slotExchange(item, true, ns)) == null) &&//slot匹配的对象返回空，slot表示的一对一的匹配。        ((Thread.interrupted() ||// 线程是否发生中断          (v = arenaExchange(item, true, ns)) == null)))//arena表示发生多线程竞争        throw new InterruptedException();//抛出中断异常    if (v == TIMED_OUT)//如果返回对象是超时        throw new TimeoutException();//抛出超时异常    return (v == NULL_ITEM) ? null : (V)v;//返回匹配后并交换的数据}

slotExchange方法

private final Object slotExchange(Object item, boolean timed, long ns) {    Node p = participant.get();//获取参与者节点    Thread t = Thread.currentThread();//获取当前线程    if (t.isInterrupted()) // 判断是否发生中断        return null;//中断唤醒直接返回    for (Node q;;) {        if ((q = slot) != null) {//表示当前属于slot的匹配，也即不是多对多的情况            if (U.compareAndSwapObject(this, SLOT, q, null)) {//cas将当前slot置空                Object v = q.item;//获取节点q的item对象                q.match = item;//将item复制给节点的match，这里表示匹配成功                Thread w = q.parked;//获取当前线程阻塞在slot中等待的线程                if (w != null)//如果w线程不为空                    U.unpark(w);//唤醒w线程                return v;//返回对象V            }            // create arena on contention, but continue until slot null            //走到这里表示竞争激烈，创建arena数组            if (NCPU > 1 && bound == 0 &&//如果CPU大于1并且bound为0表示bound前置判断，为0标识没创建arena                U.compareAndSwapInt(this, BOUND, 0, SEQ))//cas设置BOUND的值为序列号，同时保证多个线程的条件下，只有一个线程可以创建成功                arena = new Node[(FULL + 2) << ASHIFT];// FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1; MMASK = 0xff ; ASHIFT = 7  full按照CPU进行取值，如果CPU大于510,直接取MMASK，否则CPU数/2。 ASHIFT的值为7。所以2的7次方为128。也即可以保证缓存行对齐的大小        }        else if (arena != null)//走到这里表示竞争激烈，升级成多对多的匹配，所以直接返回调用arenaExchange的匹配            return null; // caller must reroute to arenaExchange        else {//这里表示当前线程找不到可以匹配的，所以将自身放到slot中等待后续匹配            p.item = item;//将item放到p.item中            if (U.compareAndSwapObject(this, SLOT, null, p))//将SLOT节点置为p节点                break;//退出            p.item = null;//cas失败将item置为空,继续循环        }    }    // await release 走到这里的条件是，当前线程将自身cas放入到slot中成功后break退出上一个循环    int h = p.hash;//获取p节点的hash值    long end = timed ? System.nanoTime() + ns : 0L;//如果有超时时间，计算超时时间    int spins = (NCPU > 1) ? SPINS : 1;//SPINS = 1 << 10 默认自旋次数为1024。具体看CPU个数    Object v;    while ((v = p.match) == null) {//如果当前p.mach为空，表示没有匹配成功        if (spins > 0) {//进行自旋            h ^= h << 1; h ^= h >>> 3; h ^= h << 10;            if (h == 0)                h = SPINS | (int)t.getId();            else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)//自旋到一定次数，让出CPU执行权。                Thread.yield();        }        else if (slot != p)//如果slot不等p继续自旋，因为竞争强烈，很快就可以匹配成功            spins = SPINS;        else if (!t.isInterrupted() && arena == null &&//判断线程是否中断并且arena是否为空                 (!timed || (ns = end - System.nanoTime()) > 0L)) {//超时判断没有超时            U.putObject(t, BLOCKER, this);//将BLOCKER置为当前对象            p.parked = t;//parked为当前线程            if (slot == p)//slot等于p节点                U.park(false, ns);//阻塞等待            p.parked = null;//再次唤醒后，将p.parked的线程置为空            U.putObject(t, BLOCKER, null);//阻塞对象也置为空        }        else if (U.compareAndSwapObject(this, SLOT, p, null)) {//cas将slot的p节点置为空            v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;//如果已经超时并且不是中断唤醒，置为TIMED_OUT，否则为null            break;//退出循环        }    }    U.putOrderedObject(p, MATCH, null);//将p.macth置为空    p.item = null;//item也置为空    p.hash = h;//p.hash位置h    return v;//所以这里返回只有两种情况：1.slot中匹配成功 2.超时后返回null}

arenaExchange方法

private final Object arenaExchange(Object item, boolean timed, long ns) {    Node[] a = arena;//获取到arena的数组    Node p = participant.get();//获取到节点p    for (int i = p.index;;) { //获取index的下标        int b, m, c; long j;                       // j is raw array offset        Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);//获取基地址+128缓存行对齐的节点Q        if (q != null && U.compareAndSwapObject(a, j, q, null)) {//如果q不为空并且cas将其职位空            Object v = q.item; //将q.item赋值给V            q.match = item;//将要交换的item值赋值为q.match            Thread w = q.parked;//获取到当前阻塞的线程            if (w != null)//如果线程不为空                U.unpark(w);//唤醒当前线程            return v;//返回匹配成功后交换过来的值        }        else if (i <= (m = (b = bound) & MMASK) && q == null) {//如果index小于最大的arena的下标，也即属于合法下标，并且当前q为空,q为空表示我自己是第一个进来的。将我自己放到arena的j下标处            p.item = item; //将当前item赋值为节点p            if (U.compareAndSwapObject(a, j, null, p)) {//cas将arena中下标为j的位置置为p节点                long end = (timed && m == 0) ? System.nanoTime() + ns : 0L;//计算超时时间                Thread t = Thread.currentThread(); // 获取当前线程                for (int h = p.hash, spins = SPINS;;) {//进行自旋                    Object v = p.match;//获取当前p.match的值                    if (v != null) {//如果匹配的值不为空，代表匹配成功了                        U.putOrderedObject(p, MATCH, null);//将p节点的match值为空                        p.item = null;//清空当前p.item的值                        p.hash = h;//将h赋值为p                        return v;返回匹配成功后交换过来的值                    }                    else if (spins > 0) {//匹配不成功，进行自旋等待                        h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift                        if (h == 0)                // initialize hash                            h = SPINS | (int)t.getId();                        else if (h < 0 &&          // approx 50% true                                 (--spins & ((SPINS >>> 1) - 1)) == 0)//h<0并且自旋计算后未0,                            Thread.yield(); //让出cpu执行权                    }                    else if (U.getObjectVolatile(a, j) != p)//获取当前arena中j下标的值不等于p                        spins = SPINS;       // 未匹配成功，赋值自旋次数                    else if (!t.isInterrupted() && m == 0 &&//判断是否中断，并且arena                             (!timed ||//是否超时等待                              (ns = end - System.nanoTime()) > 0L)) {//超时时间是否大于0                        U.putObject(t, BLOCKER, this); // 设置blocker阻塞对象为当前对象                        p.parked = t;//parked的阻塞线程为当前线程                        if (U.getObjectVolatile(a, j) == p)//判断当前arena的下标是否为p                            U.park(false, ns);//阻塞等待                        p.parked = null;//唤醒后将parked置为空                        U.putObject(t, BLOCKER, null);//唤醒后将BLOCKER置为空                    }                    else if (U.getObjectVolatile(a, j) == p &&//如果当前arena的j下标处为p                             U.compareAndSwapObject(a, j, p, null)) {//cas将j下标的p节点置为空                        if (m != 0) //m表示最大的arena数组不等于0                            U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1);//将数组缩小                        p.item = null;//将item置为空                        p.hash = h;//h赋值给p.hash                        i = p.index >>>= 1;//将index下标右移1位                        if (Thread.interrupted())//判断是否是中断唤醒                            return null;//返回空                        if (timed && m == 0 && ns <= 0L)//如果超时设置并且m为0,ns超时时间小于0表示已经超时                            return TIMED_OUT;//返回TIMED_OUT                        break;                     // expired; restart                    }                }            }            else                p.item = null; //cas失败，将p.item还原        }        else {//进入这里的情况：q不为空，代表下标竞争激烈，cas失败。所以换个下标继续循环匹配            if (p.bound != b) { //如果p.bound不等于b，表示当前b已经修改过了                p.bound = b;//重新获取最新值                p.collides = 0;//将失败次数置为0；                i = (i != m || m == 0) ? m : m - 1;//这里直接取m是因为有线程可能已经往m中放入了待匹配节点            }            else if ((c = p.collides) < m || m == FULL ||//失败次数小于m，获取m==arena最大值。m为当前的bound的值。                     !U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) {//cas将BOUND                p.collides = c + 1;//失败次数加1                i = (i == 0) ? m : i - 1; //若当前i为0，赋值m业绩最大下标,否则i的下标减1。如果是m==FULL条件进来，表示已经在最大下标处，所以从后往前查找            }            else                i = m + 1; //当cas设置bound成功后将i下标置为m+1,将cas的下标尽量打散。            p.index = i;//获取新的下标继续循环        }    }}

4.留言

​到了这里，并发工具包常用的原子性工具类已经结束了，LockSupport由于直接调用底层的park方法，较为复杂，设计到JVM的源码，暂时能力有限，后续如果看懂了在进行更新

标签: