【多线程】阻塞线程| 一图看懂ArrayBlockingQueue源码-阿里云开发者社区

【多线程】阻塞线程| 一图看懂ArrayBlockingQueue源码

2022-11-28 79

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 是一个数组实现的环形队列，经常会使用并发容器用于存储多线程间的共享数据，这样不仅可以保证线程安全，还可以简化各个线程操作

ArrayBlockingQueue简介

是一个数组实现的环形队列，经常会使用并发容器用于存储多线程间的共享数据，这样不仅可以保证线程安全，还可以简化各个线程操作

ArrayBlockingQueue队列的原理

利用了Lock锁的Condition通知机制进行阻塞控制。

核心：一把锁，两个条件

//数据元素数组
final Object[] items;
//下一个待取出元素索引
int takeIndex;
//下一个待添加元素索引
int putIndex;
//元素个数
int count;
//内部锁
final ReentrantLock lock;
//消费者
private final Condition notEmpty;
//生产者
private final Condition notFull;  
public ArrayBlockingQueue(int capacity, boolean fair) {
    ...
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

一图看懂put与take源码

网络异常，图片无法展示

1.1 put总结

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

我们在对流程图进行文字的总结

拿到线程竞争lock锁，拿到了lock锁的线程进入下一步，没有拿到lock锁的线程自旋竞争锁。
判断阻塞队列是否满了，如果满了，则调用await方法阻塞这个线程，notFull（生产者）挂起,最后释放lock锁,等待被消费者线程唤醒。
如果没有满，则调用enqueue方法将元素put进阻塞队列。
唤醒一个标记为notEmpty（消费者）的线程。

1.2 take总结

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() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        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;
    }

总结