java并发之ArrayBlockingQueue详细介绍
ArrayBlockingQueue是常用的线程集合,在线程池中也常常被当做任务队列来使用。使用频率特别高。他是维护的是一个循环队列(基于数组实现),循环结构在数据结构中比较常见,但是在源码实现中还是比较少见的。
线程安全的实现
线程安全队列,基本是离不开锁的。ArrayBlockingQueue使用的是ReentrantLock,配合两种Condition,实现了集合的线程安全操作。这里稍微说一个好习惯,下面是成员变量的声明。
1
2
3
4
5
6
7
8
9
|
private static final long serialVersionUID = -817911632652898426L;
final Object[] items;
int takeIndex;
int putIndex;
int count;
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
transient Itrs itrs = null ;
|
赋值的操作基本都是在构造函数里做的。这样有个好处,代码执行可控。成员变量的初始化也是会合并在构造方法里执行的,但是在执行顺序上需要好好斟酌,如果写在构造方法里初始化,则没有相关问题。
阻塞队列的常用场所就是生产者消费者。一般都是生产者放入,消费者从头取数据。下面重点说这两个操作。
这两个操作都是依靠锁来保证线程安全的。
生产操作
1
2
3
4
5
6
7
8
9
10
11
12
|
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();
}
}
|
put等放入操作,首先是获取锁,如果发现数据满了,就通过notFull的condition,来阻塞线程。这里的条件判定一定是用while而不是if,多线程情况下,可以被唤醒后发现又满了。
1
2
3
4
5
6
7
8
|
private void enqueue(E x) {
final Object[] items = this .items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0 ;
count++;
notEmpty.signal();
}
|
这个是入队列的操作。首先获取维护的数组。putindex就是放入操作的标志。这个操作会一直加。达到预定的长度后就变成0从头开始计数。这样插入的操作就是一个循环的操作了,count就是用来做计数的,作为能否插入数据的一个标准,插入数据后就通过notEmpty的condition发出一个信号唤醒消费线程。
消费操作
1
2
3
4
5
6
7
8
9
10
11
|
public E take() throws InterruptedException {
final ReentrantLock lock = this .lock;
lock.lockInterruptibly();
try {
while (count == 0 )
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
|
消费的方法也是这样。先获取锁,然后进行条件判断,如果没有数据,则阻塞线程。注意点和put一样。
1
2
3
4
5
6
7
8
9
10
11
12
13
|
private E dequeue() {
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;
}
|
取数据的时候,也依靠takeIndex,这是一个标志,这个数值也会一直增加,表示取的第一个数据的位置。如果这个标志走到最后,然后变成0,从头再来。这样保证取出的数据都是fifo的顺序。删除的时候如果发现迭代中,则会修改迭代器的遍历。然后通过notFull的condition来唤醒生产线程。
移除操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
public boolean remove(Object o) {
if (o == null ) return false ;
final Object[] items = this .items;
final ReentrantLock lock = this .lock;
lock.lock();
try {
if (count > 0 ) {
final int putIndex = this .putIndex;
int i = takeIndex;
do {
if (o.equals(items[i])) {
removeAt(i);
return true ;
}
if (++i == items.length)
i = 0 ;
} while (i != putIndex);
}
return false ;
} finally {
lock.unlock();
}
}
|
对于remove操作就比较麻烦了,首先获取锁之后,把两个标志位本地化,然后找到要删除的元素的位置。调用removeAt,这里删除需要对标志位做改变。
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
|
void removeAt( final int removeIndex) {
final Object[] items = this .items;
if (removeIndex == takeIndex) {
items[takeIndex] = null ;
if (++takeIndex == items.length)
takeIndex = 0 ;
count--;
if (itrs != null )
itrs.elementDequeued();
} else {
final int putIndex = this .putIndex;
for ( int i = removeIndex;;) {
int next = i + 1 ;
if (next == items.length)
next = 0 ;
if (next != putIndex) {
items[i] = items[next];
i = next;
} else {
items[i] = null ;
this .putIndex = i;
break ;
}
}
count--;
if (itrs != null )
itrs.removedAt(removeIndex);
}
notFull.signal();
}
|
如果删除的元素是位置和takeindex一样。那就可以直接删除,然后让删除标志位向后移动。如果不是,则从删除的位置开始,进行后面向前面的数据覆盖的操作。直到遇到putindex的前一个位置。然后把那个位置的数据设置为null。并且把putindex的位置往前移动一格,正在迭代的时候要删除数据并且唤醒生产线程。
感谢阅读,希望能帮助到大家,谢谢大家对本站的支持!
原文链接:https://my.oschina.net/xpbob/blog/830366
相关文章
- 64M VPS建站:能否支持高流量网站运行? 2025-06-10
- 64M VPS建站:怎样选择合适的域名和SSL证书? 2025-06-10
- 64M VPS建站:怎样优化以提高网站加载速度? 2025-06-10
- 64M VPS建站:是否适合初学者操作和管理? 2025-06-10
- ASP.NET自助建站系统中的用户注册和登录功能定制方法 2025-06-10
- 2025-07-10 怎样使用阿里云的安全工具进行服务器漏洞扫描和修复?
- 2025-07-10 怎样使用命令行工具优化Linux云服务器的Ping性能?
- 2025-07-10 怎样使用Xshell连接华为云服务器,实现高效远程管理?
- 2025-07-10 怎样利用云服务器D盘搭建稳定、高效的网站托管环境?
- 2025-07-10 怎样使用阿里云的安全组功能来增强服务器防火墙的安全性?
快网idc优惠网
QQ交流群
-
2025-05-27 29
-
2025-06-04 82
-
2025-05-27 84
-
2025-05-29 38
-
2025-05-27 94