首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >JDK源码分析-LinkedBlockingQueue

JDK源码分析-LinkedBlockingQueue

作者头像
WriteOnRead
发布2019-08-16 10:09:06
3690
发布2019-08-16 10:09:06
举报
文章被收录于专栏:WriteOnReadWriteOnRead

概述

前文「JDK源码分析-ArrayBlockingQueue」分析了 ArrayBlockingQueue 的代码实现,LinkedBlockingQueue 也是阻塞队列的实现。与前者不同的是,后者内部是由链表实现的。

LinkedBlockingQueue 的继承结构如下:

下面分析其主要方法的代码实现。

代码分析

LinkedBlockingQueue 内部有一个嵌套类 Node,它表示链表的节点,如下:

static class Node<E> {
    E item; // 节点元素

    Node<E> next; // 后继节点
    
    Node(E x) { item = x; }
}

PS: 从 Node 定义可以看出该链表是一个单链表。

主要成员变量

// 链表的容量(若不指定则为 Integer.MAX_VALUE)
private final int capacity;

// 当前元素的数量(即链表中元素的数量)
private final AtomicInteger count = new AtomicInteger();

// 链表的头节点(节点元素为空)
transient Node<E> head;

// 链表的尾结点(节点元素为空)
private transient Node<E> last;

// take、poll 等出队操作持有的锁
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
// 出队锁的条件队列
private final Condition notEmpty = takeLock.newCondition();

// put、offer 等入队操作的锁
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
// 入队锁的条件队列
private final Condition notFull = putLock.newCondition();

构造器

LinkedBlockingQueue 有三个构造器,分别如下:

// 构造器 1:无参构造器,初始容量为 Integer.MAX_VALUE,即 2^31-1
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

// 构造器 2:指定容量的构造器
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    // 初始化链表的头尾节点
    last = head = new Node<E>(null);
}

// 构造器 3:用给定集合初始化的构造器
public LinkedBlockingQueue(Collection<? extends E> c) {
    // 调用构造器 2 进行初始化
    this(Integer.MAX_VALUE);
    final ReentrantLock putLock = this.putLock;
    putLock.lock(); // Never contended, but necessary for visibility
    try {
        int n = 0;
        for (E e : c) {
            if (e == null)
                throw new NullPointerException();
            if (n == capacity)
                throw new IllegalStateException("Queue full");
            // 将集合中的元素封装成 Node 对象,并添加到链表末尾
            enqueue(new Node<E>(e));
            ++n;
        }
        count.set(n);
    } finally {
        putLock.unlock();
    }
}

enqueue 方法如下:

// 将 node 节点添加到链表末尾
private void enqueue(Node<E> node) {
    last = last.next = node;
}

主要入队方法:put(E)、offer(E, timeout, TimeUnit)、offer(E)

1. put(E) 代码如下:

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // Note: convention in all put/take/etc is to preset local var
    // holding count negative to indicate failure unless set.
    int c = -1;
    // 把 E 封装成 Node 节点
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        /*
         * Note that count is used in wait guard even though it is
         * not protected by lock. This works because count can
         * only decrease at this point (all other puts are shut
         * out by lock), and we (or some other waiting put) are
         * signalled if it ever changes from capacity. Similarly
         * for all other uses of count in other wait guards.
         */
        // 若队列已满,notFull 等待(类比生产者)
        while (count.get() == capacity) {
            notFull.await();
        }
        // node 入队
        enqueue(node);
        c = count.getAndIncrement();
        // 若该元素添加后,队列仍未满,唤醒一个其他生产者线程
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    // c==0 说明之前队列为空,出队线程处于等待状态,
    // 添加一个元素后,将出队线程唤醒(消费者)
    if (c == 0)
        signalNotEmpty();
}

signalNotEmpty 方法:

private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        // 唤醒 notEmpty 条件下的一个线程(消费者)
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

2. offer(E, timeout, TimeUnit):

public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
    if (e == null) throw new NullPointerException();
    long nanos = unit.toNanos(timeout);
    int c = -1;
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            // 等待超时,返回 false
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        // 入队
        enqueue(new Node<E>(e));
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
    return true;
}

该方法与 put 操作类似,不同的是 put 方法在队列满时会一直等待,而该方法有超时时间,超时后返回 false。

3. offer(E):

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    // 若队列已满,直接返回 false
    if (count.get() == capacity)
        return false;
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        // 队列未满,入队
        if (count.get() < capacity) {
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                // 队列未满,唤醒 notFull 下的线程,继续入队
                notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
    return c >= 0;
}

入队方法小结

1. put(E): 若队列已满,则等待,无返回值;

2. offer(E, timeout, TimeUnit): 与 put 方法类似,有超时等待和返回值;

3. offer(E): 立即返回,没有循环等待。

常用出队方法:take、poll(timeout, TimeUnit)、poll()、peek()

1. take():

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        // 队列为空,notEmpty 条件下的线程等待(消费者)
        while (count.get() == 0) {
            notEmpty.await();
        }
        // 从队列头部删除节点
        x = dequeue();
        c = count.getAndDecrement();
        // 若队列不为空,唤醒一个 notEmpty 条件下的线程(消费者)
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    // 队列已经不满了,唤醒 notFull 条件下的线程(生产者)
    if (c == capacity)
        signalNotFull();
    return x;
}

dequeue 方法:

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

signalNotFull:

private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        // 唤醒生产者
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

2. poll(timeout, TimeUnit):

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    E x = null;
    int c = -1;
    long nanos = unit.toNanos(timeout);
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        // 队列已空
        while (count.get() == 0) {
            // 超时返回 null
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        x = dequeue();
        // 若队列不空,唤醒一个 notEmpty 条件下的线程(消费者)
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    // 队列不满,唤醒 notFull 条件下的线程(生产者)
    if (c == capacity)
        signalNotFull();
    return x;
}

与 take 方法类似,多了超时等待。

3. poll():

public E poll() {
    final AtomicInteger count = this.count;
    // 队列为空,返回 null
    if (count.get() == 0)
        return null;
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        // 队列不为空,出队
        if (count.get() > 0) {
            x = dequeue();
            c = count.getAndDecrement();
            // 该元素出队后,队列仍不为空,唤醒其他消费者
            if (c > 1)
                notEmpty.signal();
        }
    } finally {
        takeLock.unlock();
    }
    // 队列已经不满,唤醒生产者
    if (c == capacity)
        signalNotFull();
    return x;
}

4. peek()

public E peek() {
    if (count.get() == 0)
        return null;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        // 头节点的后继节点
        Node<E> first = head.next;
        if (first == null)
            return null;
        else
            return first.item;
    } finally {
        takeLock.unlock();
    }
}

peek() 方法只返回头节点,并不删除。严格来说该方法并不属于出队操作,只是查询。

出队方法小结

1. take(): 获取队列头部元素,并将其移除,队列为空时阻塞等待;

2. poll(long, unit): 获取队列头部元素,并将其移除,队列为空时等待一段时间,若超时返回 null;

3. poll(): 获取队列头部元素,并将其移除,队列为空时返回 null;

小结

1. LinkedBlockingQueue 是基于单链表的阻塞队列实现,它在初始化时可以指定容量,若未指定,则默认容量为 Integer.MAX_VALUE;

2. 内部使用了 ReentrantLock 保证线程安全;

3. 常用方法:

入队:put, offer

出队:take, poll, peek

LinkedBlockingQueue 与 ArrayBlockingQueue 比较:

1. ArrayBlockingQueue 使用单个锁,可以指定是否公平;而 LinkedBlockingQueue 内部使用了两个锁:putLock 和 takeLock,都是非公平锁。

2. 入队出队区别

入队时,LinkedBlockingQueue 会判断当前元素入队后,队列是否已满,若未满,则唤醒其他生产者线程;而入队后,当队列之前为空时才唤醒其他消费者线程。ArrayBlockingQueue 则是每次入队都会唤醒消费者线程。

出队时,LinkedBlockingQueue 会判断当前元素出队后,队列是否已空,若未空,则唤醒其他消费者线程;而出队后,当队列之前为满时才唤醒其他生产者线程。ArrayBlockingQueue 则是每次出队都会唤醒生产者线程。

Stay hungry, stay foolish.

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2019-08-02,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 WriteOnRead 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
腾讯云代码分析
腾讯云代码分析(内部代号CodeDog)是集众多代码分析工具的云原生、分布式、高性能的代码综合分析跟踪管理平台,其主要功能是持续跟踪分析代码,观测项目代码质量,支撑团队传承代码文化。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档