Java-多线程-条件队列

一、队列Queue类型

JUC包中队列Queue是用于存储线程任务，常见的Queue类型有ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue和DelayQueue。

ArrayBlockingQueue

由数组组成的有界队列，队列基于数组实现,容量大小在创建ArrayBlockingQueue对象时已定义好，不可扩容。

LinkedBlockingQueue

由链接节点组成的可选有界队列，队列基于数组实现,容量大小在创建LinkedBlockingQueue对象时已定义好，不可扩容。

PriorityBlockingQueue

由优先级堆组成的无界优先级队列，内部线程是阻塞的，使用必须实现compareTo方法，这里的无界是理论上的。

DelayQueue

由优先级堆支持的、基于时间的调度队列，由优先级堆支持的、基于时间的调度队列，内部基于无界队列PriorityQueue实现，而无界队列基于数组的扩容实现。入队的对象必须要实现Delayed接口。

public static void main(String[] args) {
    DelayQueue<MovieTiket> delayQueue = new DelayQueue<MovieTiket>();
    MovieTiket tiket = new MovieTiket("电影票0",10000);
    delayQueue.put(tiket);
    MovieTiket tiket1 = new MovieTiket("电影票1",5000);
    delayQueue.put(tiket1);
    MovieTiket tiket2 = new MovieTiket("电影票2",8000);
    delayQueue.put(tiket2);
    System.out.println("message:--->入队完毕");

    while( delayQueue.size() > 0 ){
        try {
            tiket = delayQueue.take();
            System.out.println("电影票出队:"+tiket.getMsg());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class MovieTiket implements Delayed {
    //延迟时间
    private final long delay;
    //到期时间
    private final long expire;
    //数据
    private final String msg;
    //创建时间
    private final long now;

    public MovieTiket(String msg , long delay) {
        this.delay = delay;
        this.msg = msg;
        expire = System.currentTimeMillis() + delay;    //到期时间 = 当前时间+延迟时间
        now = System.currentTimeMillis();
    }

    /**
     * 用于延迟队列内部比较排序  当前时间的延迟时间 - 比较对象的延迟时间
     * 越早过期的时间在队列中越靠前
     * @param delayed
     * @return
     */
    public int compareTo(Delayed delayed) {
        return (int) (this.getDelay(TimeUnit.MILLISECONDS)
            - delayed.getDelay(TimeUnit.MILLISECONDS));
    }
}

二、Condition队列

BlockingQueue底层都是基于ReentrantLock 与 Condition队列实现的，这也是Condition队列只能在独占模式下使用的原因。
多线程下的BlockingQueue是怎样操作的？

image.png

若BlockingQueue初始容量为1，T1 T2 同时操作BlockingQueue，T1不断往BlockingQueue放，T2不断从BlockingQueue 取。T1 在放完第一次后，BlockingQueue已满，无法继续放了，T1阻塞，T2唤醒，直到BlockingQueue容量为0；T2从BlockingQueue取出，取完第一次后，BlockingQueue为空，无法继续取了，T2阻塞，T1唤醒，直到BlockingQueue容量为1。
条件关键：BlockingQueue的容量。

image.png

BlockingQueue的创建会创建一个ReentrantLock、两个队列（NotFull与NotEmpty）。

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();
}

ReentrantLock中用于多线程的同步操作，NotFull与NotEmpty用于存储不同条件的线程任务。ArrayBlockingQueue的put 和 take操作都需要配合ReentrantLock的使用，只有线程获取到锁的线程才能执行put 和 take操作。
注意1：BlockingQueue put操作是通过获取ReentrantLock锁，进入条件队列而不是CLH同步队列，同时线程会调用await方法进入阻塞状态。
注意2：条件队列中阻塞线程不会被唤醒，只有把条件队列中的线程移到CLH同步队列中才会被唤醒。

image.png

三、HashMap线程不安全

死锁

Java 1.7 HashMap会产生死锁，其数据结构为数组+链表。在多线程场景下，扩容期间存在节点位置互换指针引用的问题有可能导致闭环，主要原因还是链表中节点在扩容的时候位置发生了变化。
Java 1.8 HashMap后不会产生死锁，其数据结构为数组+红黑树。

数据丢失

HashMap的put操作在多线程下有可能产生相同的hashcode，从而造成数据覆盖，进而造成数据丢失。Java 1.7 1.8 HashMap都会产生这样的问题。