多线程系列：阻塞队列BlockingQueue分析

た入场券 2022-11-15 11:41 123阅读 0赞

### BlockingQueue ###

*  什么是BlockingQueue
 *   *  如何保证线程安全
     *   *  获取Unsafe类
 *  阻塞队列常用API
 *   *  添加元素
     *   *  offer(Object)
         *  offer(Object, Time, TimeUnit)
         *  add(Object)
         *  put(Object)
     *  移除元素
     *   *  poll()
         *  poll(Time, TimeUnit)
         *  take()
 *  常见阻塞队列的实现
 *   *  DelayQueue
     *  ArrayBlockingQueue
     *  LinkedBlockingQueue

# 什么是BlockingQueue #

BlockingQueue线程安全(AQS)的阻塞对象，在任何时候都保证了只有一个线程能访问这个队列（即数据共享）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xqY2MxMjI_size_16_color_FFFFFF_t_70]

## 如何保证线程安全 ##

通过ReentrantLock（自旋，CLH），CAS(CompareAndSwap)jvm依赖汇编指令，保证任何时候只有一个线程能操作（添加、访问元素）队列。  
Unsafe(魔法类)不安全的可以直接操作堆外内存  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xqY2MxMjI_size_16_color_FFFFFF_t_70 1]

### 获取Unsafe类 ###

public static Unsafe unsafe ;
    //通过反射的方式获取Unsafe类的实例
    public static Unsafe getUnsafe() throws IllegalAccessException { 
    	//获取单列对象unsafe
        Field unsafeField = Unsafe.class.getDeclaredField("theUnsafe");
        unsafeField.setAccessible(true);
        Unsafe unsafe = (Unsafe) unsafeField.get(null);
        return unsafe;
    }
    
    private static  long ageOffset;
    static { 
        try{ 
            unsafe = getUnsafe();
            //获取偏移量
            ageOffset = unsafe.objectFieldOffset(User.class.getField("age"));
        } catch (Exception e) { 
            System.out.println(e.getMessage());
        }
    }

# 阻塞队列常用API #

## 添加元素 ##

### offer(Object) ###

添加元素，添加成功返回true，添加失败返回false，元素为抛出异常

public boolean offer(E e) { 
     	checkNotNull(e);//检查元素是否为空 空抛出异常
        final ReentrantLock lock = this.lock;
        lock.lock();//加锁只能有一个线程进入下面的业务代码
        try { 
            if (count == items.length)//对满返回false
                return false;
            else { 
                enqueue(e);
                return true;
            }
        } finally { 
            lock.unlock();//释放锁
        }
    }
    //==============添加元素 计数器++
    private void enqueue(E x) { 
            // assert lock.getHoldCount() == 1;
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            notEmpty.signal();
        }

### offer(Object, Time, TimeUnit) ###

向队列尾部插入元素，在指定时间内未插入成功返回false，反之返回成功，空元素抛出异常。

public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException { 
            checkNotNull(e);//空判断
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
                while (count == items.length) { //队满判断
                    if (nanos <= 0) return false;
                    nanos = notFull.awaitNanos(nanos);//阻塞等待直到到达指定时间
                }
                enqueue(e);//未满直接插入
                return true;
            } finally { 
                lock.unlock();
            }
        }

### add(Object) ###

向队列的队尾插入元素，如果已满直接抛出异常，反之返回ture;

//源码
    public boolean add(E e) { 
        if (offer(e))
             return true;
         else
             throw new IllegalStateException("Queue full");
     }

### put(Object) ###

向队尾插入元素，队列已满一直等待空闲时插入，无返回值。

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

## 移除元素 ##

### poll() ###

删除并返回队首元素，队列为空返回null

public E poll() { 
            final ReentrantLock lock = this.lock;//锁对象
            lock.lock();//获取锁
            try { 
                return (count == 0) ? null : dequeue();
            } finally { 
                lock.unlock();
            }
        }

### poll(Time, TimeUnit) ###

删除并返回队首元素，为空时等待指定的时间内，如果有元素则删除返回该队首元素，反之返回null。道理同offer(Object, Time, TimeUnit)

### take() ###

删除并返回队首元素，为空时一直等到，知道队列不为空。

public E take() throws InterruptedException { 
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
                while (count == 0)
                    notEmpty.await();
                return dequeue();
            } finally { 
                lock.unlock();
            }
        }

# 常见阻塞队列的实现 #

## DelayQueue ##

按照一定数据结构（Delayed）的无界队列，按照自定义时间排序（Delayed继承了Comparable对象），只有当时间小于0时才能被取出。

public class Test2_DelayQueue { 
        static class CustomDelay implements Delayed{ 
            long order;
            long expire;
    
            public CustomDelay(int order) { 
                this.order = order;
                this.expire = order+System.currentTimeMillis();
            }
            @Override
            public long getDelay(TimeUnit unit) { //判断元素的先后取出 只有当时间小于0的才能取出
                return unit.convert(expire-System.currentTimeMillis(), TimeUnit.MILLISECONDS);
            }
            @Override
            public int compareTo(Delayed o) { //自定义排序方式
                return (int) (getDelay(TimeUnit.MILLISECONDS)- o.getDelay(TimeUnit.MILLISECONDS));
            }
        }
        public static void main(String[] args) throws InterruptedException { 
            DelayQueue<CustomDelay> delayQueue = new DelayQueue<>();
            //按升序输出
            delayQueue.put(new CustomDelay(5));
            delayQueue.put(new CustomDelay(4));
            delayQueue.put(new CustomDelay(7));
            delayQueue.put(new CustomDelay(1));
            while (delayQueue.size()>0) { 
                System.out.println(delayQueue.take().order);
            }
        }
    }

## ArrayBlockingQueue ##

数组节点支持的右键队列。

## LinkedBlockingQueue ##

链表节点支持的有界队列。

--------------------

待补充点ReentrantLock,AQS

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xqY2MxMjI_size_16_color_FFFFFF_t_70]: /images/20221022/6abe8beff97c462f89b7c7c1bb626474.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xqY2MxMjI_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210331191845588.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xqY2MxMjI=,size_16,color_FFFFFF,t_70