详细分析Java并发集合LinkedBlockingQueue的用法

时间:2022-12-10 09:47:49

在上一章我们讲解了arrayblockingqueue,用数组形式实现的阻塞队列。

数组的长度在创建时就必须确定,如果数组长度小了,那么arrayblockingqueue队列很容易就被阻塞,如果数组长度大了,就容易浪费内存。

而队列这个数据结构天然适合用链表这个形式,而linkedblockingqueue就是使用链表方式实现的阻塞队列。

一. 链表实现

1.1 node内部类

?
1
2
3
4
5
6
7
8
9
10
11
/**
 * 链表的节点,同时也是通过它来实现一个单向链表
 */
static class node<e> {
  e item;
 
  // 指向链表的下一个节点
  node<e> next;
 
  node(e x) { item = x; }
}

有一个变量e储存数据,有一个变量next指向下一个节点的引用。它可以实现最简单地单向列表。

1.2 怎样实现链表

?
1
2
3
4
5
6
7
8
9
/**
 * 它的next指向队列头,这个节点不存储数据
 */
transient node<e> head;
 
/**
 * 队列尾节点
 */
private transient node<e> last;

要实现链表,必须有两个变量,一个表示链表头head,一个表示链表尾last。head和last都会在linkedblockingqueue对象创建的时候被初始化。

?
1
last = head = new node<e>(null);

注意这里用了一个小技巧,链表头head节点并没有存放数据,它指向的下一个节点,才真正存储了链表中第一个数据。而链表尾last的确储存了链表最后一个数据。

1.3 插入和删除节点

?
1
2
3
4
5
6
7
8
9
/**
  * 向队列尾插入节点
  */
 private void enqueue(node<e> node) {
   // assert putlock.isheldbycurrentthread(); // 当前线程肯定获取了putlock锁
   // 将原队列尾节点的next引用指向新节点node,然后再将node节点设置成队列尾节点last
   // 这样就实现了向队列尾插入节点
   last = last.next = node;
 }

在链表尾插入节点很简单,将原队列尾last的下一个节点next指向新节点node,再将新节点node赋值给队列尾last节点。这样就实现了插入一个新节点。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 移除队列头节点,并返回被删除的节点数据
private e dequeue() {
  // assert takelock.isheldbycurrentthread(); // 当前线程肯定获取了takelock锁
  // assert head.item == null;
 
  node<e> h = head;
  // first节点中才存储了队列中第一个元素的数据
  node<e> first = h.next;
  h.next = h; // help gc
  // 设置新的head值,相当于删除了first节点。因为head节点本身不储存数据
  head = first;
  // 队列头的数据
  e x = first.item;
  // 移除原先的数据
  first.item = null;
  return x;
}

要注意head并不是链表头,它的next才是指向链表头,所以删除链表头也很简单,就是将head.next赋值给head,然后返回原先head.next节点的数据。

删除的时候,就要注意链表为空的情况。head.next的值使用enqueue方法添加的。当head.next==last的时候,表示已经删除到最后一个元素了,当head.next==null的时候,就不能删除了,因为链表已经为空了。这里没有做判断,是因为在调用dequeue方法的地方已经做过判断了。

二. 同步锁reentrantlock和条件condition

因为阻塞队列在队列为空和队列已满的情况下,都必须阻塞等待,那么就天然需要两个条件。而为了保证多线程并发安全,又需要一个同步锁。这个在arrayblockingqueue中已经说过了。

这里我们来说说linkedblockingqueue不一样的地方。

?
1
2
3
4
5
6
7
8
9
10
11
/** 独占锁,用于处理插入队列操作的并发问题,即put与offer操作 */
private final reentrantlock putlock = new reentrantlock();
 
/** 队列不满的条件condition,它是由putlock锁生成的 */
private final condition notfull = putlock.newcondition();
 
/** 独占锁,用于处理删除队列头操作的并发问题,即take与poll操作 */
private final reentrantlock takelock = new reentrantlock();
 
/** 队列不为空的条件condition, 它使用takelock锁生成的 */
private final condition notempty = takelock.newcondition();

2.1 putlock与takelock

我们发现使用了两把锁:

  1. putlock 同步所有插入元素的操作,即put与offer系列方法的操作。
  2. takelock 同步删除和获取元素的操作,即take与poll系列方法操作。

按照上面的说法,可能会出现同时插入元素和删除元素的操作,那么就不会出现问题么?

我们来具体分析一个下,对于队列来说操作分为三种:

  1. 在队列尾插入元素。即put与offer系列方法,它们会涉及两个变量,队列中元素个数count,和队列尾节点last。(不会涉及head节点)
  2. 移除队列头元素。即take与poll系列方法,它们会涉及两个变量,队列中元素个数count,和head节点。(不会涉及队列尾节点last)
  3. 查看队列头元素。即 element()与peek()方法,它们会涉及两个变量,队列中元素个数count,和head节点。(不会涉及队列尾节点last)

因此使用putlock锁来保持last变量的安全,使用takelock锁来保持head变量的安全。

对于都涉及了队列中元素个数count变量,所以使用atomicinteger来保证并发安全问题。

?
1
2
/** 队列中元素的个数,这里使用atomicinteger变量,保证并发安全问题 */
private final atomicinteger count = new atomicinteger();

2.2 notfull与notempty

  1. notfull 是由putlock锁生成的,因为当插入元素时,我们要判断队列是不是已满。
  2. notempty 是由takelock锁生成的,因为当删除元素时,我们要判断队列是不是为空。

2.3 控制流程

当插入元素时:

  1. 先使用putlock.lockinterruptibly()保证只有一个线程进行插入操作.
  2. 然后利用count变量,判断队列是否已满.
  3. 满了就调用notfull.await()方法,让当前线程等待。因为notfull是由putlock产生的,这里已经获取到锁了,所以可以调用await方法。
  4. 没满就调用 enqueue方法,向队列尾插入新元素。
  5. 调用count.getandincrement()方法,将队列中元素个数加一,并保证多线程并发安全。
  6. 调用signalnotempty方法,唤醒正在等待获取元素的线程。

当删除元素时:

  1. 先使用takelock.lockinterruptibly()保证只有一个线程进行删除操作.
  2. 然后利用count变量,判断队列是否为空.
  3. 队列为空就调用notempty.await()方法,让当前线程等待。因为notempty是由takelock产生的,这里已经获取到锁了,所以可以调用await方法。
  4. 没满就调用 dequeue方法,删除队列头元素。
  5. 调用count.getanddecrement()方法,将队列中元素个数减一,并保证多线程并发安全。
  6. 调用signalnotfull方法,唤醒正在等待插入元素的线程。

还要注意一下,condition的signal和await方法必须在获取锁的情况下调用。因此就有了signalnotempty和signalnotfull方法:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
 * 唤醒在notempty条件下等待的线程,即移除队列头时,发现队列为空而*等待的线程。
 * 注意,因为要调用condition的signal方法,必须获取对应的锁,所以这里调用了takelock.lock()方法。
 * 当队列中插入元素(即put或offer操作),那么队列肯定不为空,就会调用这个方法。
 */
private void signalnotempty() {
  final reentrantlock takelock = this.takelock;
  takelock.lock();
  try {
    notempty.signal();
  } finally {
    takelock.unlock();
  }
}
 
/**
 * 唤醒在notfull条件下等待的线程,即队列尾添加元素时,发现队列已满而*等待的线程。
 * 注意,因为要调用condition的signal方法,必须获取对应的锁,所以这里调用了putlock.lock()方法
 * 当队列中删除元素(即take或poll操作),那么队列肯定不满,就会调用这个方法。
 */
private void signalnotfull() {
  final reentrantlock putlock = this.putlock;
  putlock.lock();
  try {
    notfull.signal();
  } finally {
    putlock.unlock();
  }
}

三. 插入元素方法

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public void put(e e) throws interruptedexception {
  if (e == null) throw new nullpointerexception();
  // 记录插入操作前元素的个数
  int c = -1;
  // 创建新节点node
  node<e> node = new node<e>(e);
  final reentrantlock putlock = this.putlock;
  final atomicinteger count = this.count;
  putlock.lockinterruptibly();
  try {
    //表示队列已满,那么就要调用notfull.await方法,让当前线程等待
    while (count.get() == capacity) {
      notfull.await();
    }
    // 向队列尾插入新元素
    enqueue(node);
    // 将当前队列元素个数加1,并返回加1之前的元素个数。
    c = count.getandincrement();
    // c + 1 < capacity表示队列未满,就唤醒可能等待插入操作的线程
    if (c + 1 < capacity)
      notfull.signal();
  } finally {
    putlock.unlock();
  }
  // c == 0表示插入之前,队列是空的。队列从空到放入一个元素时,
  // 才唤醒等待删除的线程
  // 防止频繁获取takelock锁,消耗性能
  if (c == 0)
    signalnotempty();
}

以put方法为例,大体流程与我们前面介绍一样,这里有一个非常怪异的代码,当插入完元素时,如果发现队列未满,那么调用notfull.signal()唤醒等待插入的线程。

大家就很疑惑了,一般来说,这个方法应该放在删除元素(take系列的方法里),因为当我们删除一个元素,那么队列肯定是未满的,那么调用notfull.signal()方法,唤醒等待插入的线程。

这里这么做主要是因为调用signal方法,必须先获取对应的锁,而在take系列的方法里使用的锁是takelock,那么想调用notfull.signal()方法,必须先获取putlock锁,这样的话会性能就会下降,所以用了另一种方式。

  1. 首先我们应该知道signal方法,当有线程在这个条件下等待时,才会唤醒其中一个线程,当没有线程等待时,这个方法相当于什么都没做。所以这个方法的意义是可能会唤醒等待的一个线程。
  2. 当队列未满时,我们都调用notfull.signal()尝试去唤醒一个等待插入线程。而且这里已经获取putlock锁了,所以不耗时。
  3. 但是有一个问题,当队列已满的时候,所有插入操作的线程,都会等待,就没有机会调用notfull.signal()方法,那么唤醒这些等待线程呢?
  4. 唤醒这些线程的启动条件,必须是由删除元素操作触发的,因为只有删除队列才会不满。因为在take方法中 if (c == capacity) signalnotfull();

四. 删除队列头元素

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public e take() throws interruptedexception {
  e x;
  int c = -1;
  final atomicinteger count = this.count;
  final reentrantlock takelock = this.takelock;
  takelock.lockinterruptibly();
  try {
    //表示队列为空,那么就要调用notempty.await方法,让当前线程等待
    while (count.get() == 0) {
      notempty.await();
    }
    // 删除队列头元素,并返回它
    x = dequeue();
    // 返回当前队列个数,然后将队列个数减一
    c = count.getanddecrement();
    // c > 1表示队列不为空,就唤醒可能等待删除操作的线程
    if (c > 1)
      notempty.signal();
  } finally {
    takelock.unlock();
  }
  /**
   * c == capacity表示删除操作之前,队列是满的。只有从满队列中删除一个元素时,
   * 才唤醒等待插入的线程
   * 防止频繁获取putlock锁,消耗性能
   */
  if (c == capacity)
    signalnotfull();
  return x;
}

为什么调用notempty.signal()方法原因,对比一下我们在插入元素方法中的解释。

五. 查看队列头元素

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 查看队列头元素
public e peek() {
  // 队列为空,返回null
  if (count.get() == 0)
    return null;
  final reentrantlock takelock = this.takelock;
  takelock.lock();
  try {
    // 获取队列头节点first
    node<e> first = head.next;
    // first == null表示队列为空,返回null
    if (first == null)
      return null;
    else
      // 返回队列头元素
      return first.item;
  } finally {
    takelock.unlock();
  }
}

查看队列头元素,涉及到head节点,所以必须使用takelock锁。

六. 其他重要方法

6.1 remove(object o)方法

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 从队列中删除指定元素o
public boolean remove(object o) {
  if (o == null) return false;
  // 因为不是删除列表头元素,所以就涉及到head和last两个变量,
  // putlock与takelock都要加锁
  fullylock();
  try {
    // 遍历整个队列,p表示当前节点,trail表示当前节点的前一个节点
    // 因为是单向链表,所以需要记录两个节点
    for (node<e> trail = head, p = trail.next;
       p != null;
       trail = p, p = p.next) {
      // 如果找到了指定元素,那么删除节点p
      if (o.equals(p.item)) {
        unlink(p, trail);
        return true;
      }
    }
    return false;
  } finally {
    fullyunlock();
  }
}

从列表中删除指定元素,因为删除的元素不一定在列表头,所以可能会head和last两个变量,所以必须同时使用putlock与takelock两把锁。因为是单向链表,需要一个辅助变量trail来记录前一个节点,这样才能删除当前节点p。

6.2 unlink(node<e> p, node<e> trail) 方法

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 删除当前节点p,trail代表p的前一个节点
void unlink(node<e> p, node<e> trail) {
  // 将当前节点的数据设置为null
  p.item = null;
  // 这样就在链表中删除了节点p
  trail.next = p.next;
  // 如果节点p是队列尾last,而它被删除了,那么就要将trail设置为last
  if (last == p)
    last = trail;
  // 将元素个数减一,如果原队列是满的,那么就调用notfull.signal()方法
  // 其实这个不用判断直接调用的,因为这里肯定获取了putlock锁
  if (count.getanddecrement() == capacity)
    notfull.signal();
}

要在链表中删除一个节点p,只需要将p的前一个节点trail的next指向节点p的下一个节点next。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持服务器之家。

原文链接:https://www.jianshu.com/p/fb79f074be28