首页 > 代码库 > Java并发包源码分析

Java并发包源码分析

2024-08-20 16:27:45   225人阅读

　　并发是一种能并行运行多个程序或并行运行一个程序中多个部分的能力。如果程序中一个耗时的任务能以异步或并行的方式运行，那么整个程序的吞吐量和可交互性将大大改善。现代的PC都有多个CPU或一个CPU中有多个核,是否能合理运用多核的能力将成为一个大规模应用程序的关键。

　　Java基础部分知识总结点击Java并发基础总结。Java多线程相关类的实现都在Java的并发包concurrent，concurrent包主要包含3部分内容，第一个是atomic包，里面主要是一些原子类，比如AtomicInteger、AtomicIntegerArray等；第二个是locks包，里面主要是锁相关的类，比如ReentrantLock、Condition等；第三个就是属于concurrent包的内容，主要包括线程池相关类（Executors）、阻塞集合类（BlockingQueue）、并发Map类（ConcurrentHashMap）、线程相关类（Thread、Runnable、Callable）等。

技术分享

atomic包源码分析

　　atomic包是专门为线程安全设计的Java包，包含多个原子操作类。其基本思想就是在多线程环境下，当有多个线程同时执行这些类的实例的方法时，具有排他性，一个线程进入方法执行指令时，不会被其他的线程打断，而别的线程就像自旋锁一样，一直等待该方法执行完成。

　　原子变量的底层使用了处理器提供的原子指令，但是不同的CPU架构可能提供的原子指令不一样，也有可能需要某种形式的内部锁，所以该方法不能绝对保证线程不被阻塞。

　　atomic包一共有12个类，四种原子更新方式，分别是原子更新基本类型、原子更新数组、原子更新引用和原子更新字段。JDK1.5中引入了底层的支持，在int、long和对象的引用等类型上都公开了CAS的操作，并且JVM把它们编译为底层硬件提供的最有效的方法，在运行CAS的平台上，运行时把它们编译为相应的机器指令。在java.util.concurrent.atomic包下面的所有的原子变量类型中，比如AtomicInteger，都使用了这些底层的JVM支持为数字类型的引用类型提供一种高效的CAS操作。

　　Unsafe中的操作一般都是基于CAS来实现的，CAS就是Compare and Swap的意思，比较并操作。很多的cpu直接支持CAS指令。CAS是一项乐观锁技术，当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。CAS有3个操作数，内存值V，旧的预期值A，要修改的新值B。当且仅当预期值A和内存值V相同时，将内存值V修改为B，否则什么都不做。

/**
 * AtomicMain
 * atomic class test
 */
public class AtomicMain {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService   executor  = Executors.newCachedThreadPool();
        AtomicInteger     data      = new AtomicInteger(0);
        AtomicIntegerArray array    = new AtomicIntegerArray(10);
        AtomicReference   reference = new AtomicReference();

        /* AtomicInteger测试 */
        executor.execute(new AtomicIntegerTask(data));
        executor.execute(new AtomicIntegerTask(data));

        /* AtomicIntegerArray测试 */
        executor.execute(new AtomicIntegerArrayTask(array));
        executor.execute(new AtomicIntegerArrayTask(array));

        User user = new User("xxx", 18);
        reference.set(user);
        executor.execute(new AtomicReferenceTask(reference));

        /**
         * shutdown表示线程池不再接收新的任务了，
         * 而不是阻塞到线程池任务执行完成之后再返回
         */
        executor.shutdown();
        /* 延时保证线程池任务执行完毕 */
        Thread.sleep(100);

        System.out.println(data);

        for (int i = 0; i < 10; i++) {
            System.out.print(array.get(i) + " ");
        }
        System.out.println();

        System.out.println(user);
    }

    /**
     * AtomicInteger
     */
    static class AtomicIntegerTask implements Runnable {
        private AtomicInteger data;

        public AtomicIntegerTask(AtomicInteger data) {
            this.data =http://www.mamicode.com/ data;
        }

        public void run() {
            int cnt = 10;

            while (cnt-- > 0) {
                data.incrementAndGet();
            }
        }
    }

    /**
     * 传进来的Array大小至少为10
     * AtomicIntegerArray是原子性的，保证对该array整个内存操作的原子性，
     * 也就是说不可能同时有A线程对array[0]操作，而B线程对array[1]操作
     */
    static class AtomicIntegerArrayTask implements Runnable {
        private AtomicIntegerArray array;

        public AtomicIntegerArrayTask(AtomicIntegerArray array) {
            this.array = array;
        }

        public void run() {
            int cnt = 10;

            while (cnt-- > 0) {
                for (int i = 0; i < 10; i++) {
                    array.getAndAdd(i, 1);
                }
            }
        }
    }

    static class AtomicReferenceTask implements Runnable {
        private AtomicReference reference;

        public AtomicReferenceTask(AtomicReference reference) {
            this.reference = reference;
        }

        public void run() {
            reference.set(new User("luoxn28", 23));
        }
    }

    static class User {
        public String name;
        public int    age;

        public User(String name, int age) {
            this.name = name;
            this.age  = age;
        }

        @Override
        public String toString() {
            return "User{" +
                    "name=‘" + name + ‘\‘‘ +
                    ", age=" + age +
                    ‘}‘;
        }
    }
}

AtomicInteger. incrementAndGet流程

/**
 * 原子自增1
 * this表示AtomicInteger实例
 * valueOffset表示value数据域相对于this的内存地址的偏移位置
 */
public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}
public final int getAndAddInt(Object var1, long var2, int var4) {
    int var5;
    do {
        /* 获取value在内存中的值，然后进行CAS操作 */
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

    return var5;
}

lock包源码分析

　　lock包里面主要是锁相关的类，比如ReentrantLock、Condition等。

　　Lock接口主要有lock、lockInterruptibly、tryLock、unlock、newCondition等方法：

public interface Lock {

    /**
     * 获取锁，获取不到时该线程一直处于休眠状态
     */
    void lock();

    /**
     * 如果所可用则获取锁；否则线程处理休眠状态，如果此时发生中断，则抛出InterruptException异常
     */
    void lockInterruptibly() throws InterruptedException;

    /**
     * 如果锁可用则获取锁并返回true，否则返回false
     */
    boolean tryLock();

    /**
     * tryLock的待超时时间版本
     */
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

    /**
     * 释放锁
     */
    void unlock();

    /**
     * 返回用来与此 Lock 实例一起使用的 Condition 实例
     */
    Condition newCondition();
}

使用Lock示例：

public class LockMain {
    public static void main(String[] args) throws InterruptedException {
        Lock lock = new ReentrantLock();
        AtomicInteger data = new AtomicInteger(0);

        ExecutorService executor = Executors.newCachedThreadPool();
        executor.execute(new LockAddTask(lock, data));
        executor.execute(new LockAddTask(lock, data));

        executor.shutdown();
        Thread.sleep(10);

        System.out.println(data.get());
    }

    static class LockAddTask implements Runnable {
        private Lock lock;
        private AtomicInteger data;

        public LockAddTask(Lock lock, AtomicInteger data) {
            this.lock = lock;
            this.data =http://www.mamicode.com/ data;
        }

        public void run() {
            int cnt = 10;

            while (cnt-- > 0) {
                try {
                    lock.lockInterruptibly();
                    data.getAndIncrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }
    }
}

concurrent包源码分析

BlockingQueue

public interface BlockingQueue<E> extends Queue<E> {

    /**
     * 底层调用的是offer，如果满了抛出异常
     */
    boolean add(E e);

    /**
     * 当集合为满时，一直等待
     */
    void put(E e) throws InterruptedException;

    /**
     * 当集合为满时，一直等待到超时
     */
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
    boolean offer(E e);

    /**
     * 当集合为空时，始终等待
     */
    E take() throws InterruptedException;

    /**
     * 当集合为空时，一直等到超时，如果还为空则返回null
     */
    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

    /**
     * 底层调用的是poll，如果空了抛出异常
     */
    boolean remove(Object o);

    //...
}

ArrayBlockingQueue

　　ArrayBlockingQueue是一个基于数组的有界阻塞队列，按照FIFO（先进先出）原则对元素进行排序，在构造方法中会new一个数组，并且new ReentrantLock，并且初始化notEmpty和notFull两个Condition。

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

　　执行put操作时，首先获取lock，如果数组已经满了，则调用notFull.await等待；否则调用enqueue插入元素，插入成功后把count计数值加1，调用notEmpty.signal。判断数组是否满了是根据count是否等于数组长度来确定的，因为往数组中插入元素时，首先从下标为0位置开始插入，插到下标为array.length-1时，如果count小于array.length，则下一次从下标为0位置插入。

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}
private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

　　执行take操作时，首先获取lock，如果数组为空，则调用notEmpty.await等待；否则调用dequeue取出元素，取出成功后把count计数值减1，调用notFull.signal。

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}
private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return x;
}

　　lock的wait/signal更多知识：http://www.cnblogs.com/alphablox/archive/2013/01/20/2868479.html

LinkedBlockingQueue

　　LinkedBlockingQueue是基于链表结构的阻塞队列，按照FIFO（先进先出）原则对元组进行排序，新元素是尾部插入，吞吐量通常高于ArrayBlockingQueue。该类中包含一个takeLock和基于takeLock的Condition对象notEmpty，一个putLock锁，和基于putLock的Condition对象notFull。在构造方法中会新new一个Node，last和head都指向该Node节点。

public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);
}

　　执行put操作时，首先获取putLock，如果链表节点数已经达到上限，则调用notFull.await等待；否则调用enqueue插入元素，插入成功后把count值原子加1，如果链表节点数未达到上限，则调用notFull.signal。然后获取takeLock，再调用notEmpty.signal通知。

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(node);
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}
private void enqueue(Node<E> node) {
    last = last.next = node;
}
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

　　执行take操作时，首先获取takeLock，如果链表为空，则调用notEmpty.await等待；否则调用dequeue取出元素，然后把count值原子减1，如果此时链表非空，则调用notEmpty.signal。然后获取putLock，再调用putLock.signal通知。

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        while (count.get() == 0) {
            notEmpty.await();
        }
        x = dequeue();
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}
private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

ConcurrentHashMap

       ConcurrentHashMap是concurrent包中一个重要的类，其高效支并发操作，被广泛使用，Spring框架的底层数据结构就是使用ConcurrentHashMap实现的。同HashTable相比，它的锁粒度更细，而不是像HashTable一样为每个方法都添加了synchronized锁。

       Java8中的ConcurrentHashMap废弃了Segment（锁段）的概念，而是用CAS和synchronized方法来实现。利用CAS来获取table数组中的单个Node节点，获取成功进行更新操作时，再使用synchronized处理对应Node节点所对应链表（或红黑树）中的数据。

使用ConcurrentHashMap程序示例

/**
 * HashMapMain test
 */
public class HashMapMain {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, String> hashMap = new ConcurrentHashMap<String, String>();
        ExecutorService executorService = Executors.newCachedThreadPool();

        executorService.execute(new HashMapPutTask(hashMap));
        executorService.execute(new HashMapPutTask(hashMap));
        executorService.execute(new HashMapPutTask(hashMap));

        executorService.shutdown();
        /**
         * Main thread wait for other thread over.
         */
        Thread.sleep(2000);

        Set<Map.Entry<String, String>> set = hashMap.entrySet();
        Iterator<Map.Entry<String, String>> iter = set.iterator();
        int i = 0;
        while (iter.hasNext()) {
            Map.Entry<String, String> keyValue =http://www.mamicode.com/ iter.next();
            System.out.println(++i + " -> " + keyValue.getKey() + ": " + keyValue.getValue());
        }
    }

    static class HashMapPutTask implements Runnable {
        private ConcurrentHashMap<String, String> hashMap;

        public HashMapPutTask(ConcurrentHashMap<String, String> hashMap) {
            this.hashMap = hashMap;
        }

        public void run() {
            int cnt = 10;

            while (cnt-- > 0) {
                String key   = UUID.randomUUID().toString();
                String value = UUID.randomUUID().toString();
                hashMap.put(key, value);
            }
        }
    }
}

几个核心的内部类：

Node

　　Node是最核心的内部类，它包装了key-value键值对，所有插入ConcurrentHashMap的数据都包装在这里面。它与HashMap中的定义很相似，但是但是有一些差别它对value和next属性设置了volatile同步锁，它不允许调用setValue方法直接改变Node的value域，它增加了find方法辅助map.get()方法。

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
    // ...
}

TreeNode

　　树节点类，另外一个核心的数据结构。当链表长度过长的时候，会转换为TreeNode。但是与HashMap不相同的是，它并不是直接转换为红黑树，而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的操作。而且TreeNode在ConcurrentHashMap继承自Node类，而并非HashMap中的继承自LinkedHashMap.Entry<K,V>类，也就是说TreeNode带有next指针，这样做的目的是方便基于TreeBin的访问。

TreeBin

       这个类并不负责包装用户的key、value信息，而是包装的很多TreeNode节点。它代替了TreeNode的根节点，也就是说在实际的ConcurrentHashMap“数组”中，存放的是TreeBin对象，而不是TreeNode对象，这是与HashMap的区别。另外这个类还带有了读写锁。

put操作

　　ConcurrentHashMap最常用的就是put和get两个方法。现在来介绍put方法，这个put方法依然沿用HashMap的put方法的思想，根据hash值计算这个新插入的点在table中的位置i，如果i位置是空的，直接放进去，否则进行判断，如果i位置是树节点，按照树的方式插入新的节点，否则把i插入到链表的末尾。ConcurrentHashMap中依然沿用这个思想，有一个最重要的不同点就是ConcurrentHashMap不允许key或value为null值。另外由于涉及到多线程，put方法就要复杂一点。在多线程中可能有以下两个情况

  • 如果一个或多个线程正在对ConcurrentHashMap进行扩容操作，当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到，是因为transfer方法中在空结点上插入forward节点，如果检测到需要插入的位置被forward节点占有，就帮助进行扩容；
  • 如果检测到要插入的节点是非空且不是forward节点，就对这个节点加锁，这样就保证了线程安全。尽管这个有一些影响效率，但是还是会比hashTable的synchronized要好得多。

　　整体流程就是首先定义不允许key或value为null的情况放入  对于每一个放入的值，首先利用spread方法对key的hashcode进行一次hash计算，由此来确定这个值在table中的位置。如果这个位置是空的，那么直接放入，而且不需要加锁操作。

　　如果这个位置存在结点，说明发生了hash碰撞，首先进入sychnorized同步代码块，然后判断这个节点的类型。如果是链表节点（fh>0）,则得到的结点就是hash值相同的节点组成的链表的头节点。需要依次向后遍历确定这个新加入的值所在位置。如果遇到hash值与key值都与新加入节点是一致的情况，则只需要更新value值即可。否则依次向后遍历，直到链表尾插入这个结点。  如果加入这个节点以后链表长度大于8，就把这个链表转换成红黑树。如果这个节点的类型已经是树节点的话，直接调用树节点的插入方法进行插入新的值。

public V put(K key, V value) {
    return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
        //不允许 key或value为null
    if (key == null || value =http://www.mamicode.com/= null) throw new NullPointerException();
    //计算hash值
    int hash = spread(key.hashCode());
    int binCount = 0;
    //死循环 何时插入成功 何时跳出
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //如果table为空的话，初始化table
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        //根据hash值计算出在table里面的位置 
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            //如果这个位置没有值 ，直接放进去，不需要加锁
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        //当遇到表连接点时，需要进行整合表的操作
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            //结点上锁  这里的结点可以理解为hash值相同组成的链表的头结点
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    //fh〉0 说明这个节点是一个链表的节点 不是树的节点
                    if (fh >= 0) {
                        binCount = 1;
                        //在这里遍历链表所有的结点
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            //如果hash值和key值相同  则修改对应结点的value值
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value;
                                break;
                            }
                            Node<K,V> pred = e;
                            //如果遍历到了最后一个结点，那么就证明新的节点需要插入 就把它插入在链表尾部
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);
                                break;
                            }
                        }
                    }
                    //如果这个节点是树节点，就按照树的方式插入值
                    else if (f instanceof TreeBin) {
                        Node<K,V> p;
                        binCount = 2;
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            if (binCount != 0) {
                //如果链表长度已经达到临界值8 就需要把链表转换为树结构
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //将当前ConcurrentHashMap的元素数量+1
    addCount(1L, binCount);
    return null;
}

get方法

　　get方法比较简单，给定一个key来确定value的时候，必须满足两个条件  key相同  hash值相同，对于节点可能在链表或树上的情况，需要分别去查找。

public V get(Object key) {  
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;  
    //计算hash值  
    int h = spread(key.hashCode());  
    //根据hash值确定节点位置  
    if ((tab = table) != null && (n = tab.length) > 0 &&  
        (e = tabAt(tab, (n - 1) & h)) != null) {  
        //如果搜索到的节点key与传入的key相同且不为null,直接返回这个节点    
        if ((eh = e.hash) == h) {  
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))  
                return e.val;  
        }  
        //如果eh<0 说明这个节点在树上 直接寻找  
        else if (eh < 0)  
            return (p = e.find(h, key)) != null ? p.val : null;  
         //否则遍历链表 找到对应的值并返回  
        while ((e = e.next) != null) {  
            if (e.hash == h &&  
                ((ek = e.key) == key || (ek != null && key.equals(ek))))  
                return e.val;  
        }  
    }  
    return null;  
}

 

参考：

　　1、ConcurrentHashMap源码分析（JDK8版本）

　　2、Java并发基础总结

Java并发包源码分析


联系
我们