Perl线程队列：Thread::Queue - 好文

(Thread::Queue)队列数据结构(FIFO)是线程安全的
，它保证了某些线程从一端写入数据，另一些线程从另一端读取数据。只要队列已经满了，写入操作就自动被阻塞直到有空间支持写操作，只要队列空了，读取操作就会自动阻塞直到队列中有数据可读。这种模式自身就保证了线程安全性。

创建队列

new()
new(LIST)
new()可以创建一个空队列，或者根据已有的列表创建队列，列表中的元素会按照先后顺序放进这个队列中。

哪些元素可放进队列

可以被threads::shared共享的数据都可以放进队列。包括：
Ordinary scalars Array refs Hash refs Scalar refs Objects based on the above
放进队列的数据必须是已经共享的，如果没有共享，则会自动克隆(递归克隆)一份后将其共享并将共享后放进队列。

例如，下面首先会创建一个空队列，由于@arr未共享，所以会先通过&shared([])将一个空列表(匿名列表引用)共享，并将@arr
中的3个元素放进去，然后再将这个共享的匿名列表放进队列。
my @arr = qw(foo bar baz); $q->enqueue(\@arr);
但是下面的数据是已经共享的，将会直接放进队列中，而不需要先克隆一份。
my @arr :shared = qw(foo bar baz); $q->enqueue(\@arr);
注意是整个列表引用，它将是队列中的单个元素，而不是将那三个元素放进队列。另外，需要克隆的时候，原始数据结构不会被共享，而是共享它的匿名结构的克隆。

操作队列的基本方法

enqueue(LIST)
将列表中的元素放进队列的尾部，给定的列表可以只有一个元素。默认情况下队列长度可以无限增长，但可以通过limit来设置

dequeue()
dequeue(COUNT)
从队列的头部移除COUNT(默认为1)个元素，并返回它们。如果队列中的元素数量少于请求移除的数量，则线程将被阻塞直到有足够的元素可返回

dequeue_nb()
dequeue_nb(COUNT)

以非阻塞的方式请求从队列头部移除COUNT(默认为1)个元素并返回它们，如果队列中元素个数少于请求移除的数量，则立即返回已存在的那些元素而不会阻塞等待，如果队列为空，则立即返回undef

dequque_timed(TIMEOUT)
dequque_timed(TIMEOUT, COUNT)

从队列的头部移除COUNT(默认为1)个元素，并返回它们。如果队列中的元素个数少于请求移除的个数，则阻塞直到有足够的元素可以返回或者阻塞直到超时。如果等待到了超时，则返回队列中已存在的元素，如果队列为空，则返回undef。如果省略了TIMEOUT或者定义为undef或小于等于0的值，则等价于
dequeue_nb。

timeout可以是以秒为单位计时的正数、小数(如4、2.5等)，也可以是从1970-01-01 00:00:00距离现在已过去的秒数，即epoch时间戳。

pending()
返回队列中还有多少个元素，如果队列已被结束(end())了或已经没有元素了，则返回undef。

$QUEUE->limit = N
设置队列的最大长度为N，如果N=0则表示无限制。设置了limit后，后续的enqueue
会被阻塞直到pending的数量小于limit的值，但是需要注意的是，在一次性向队列中enqueue多个元素导致跨了limit值时不会阻塞。见示例：
my $q = Thread::Queue->new(1, 2); $q->limit = 4; $q->enquque(3,4,5); # 不会阻塞
$q->enqueue(6); # 阻塞，直到队列中元素少于4 my $size = $q->limit; # 返回当前的队列限制值，可能会返回undef
$q->limit = 0; # 不再限制队列长度
end()
声明这个队列已经不会再有元素放进来了，也就是告诉对端要关闭队列，使其不要等待或阻塞。这也可以通过enqueue(undef)发送undef的方式来实现。

更多操作队列的方法

peek()
peek(INDEX)

从队列中返回指定索引位(注意：从队列头部开始计数)的元素但却不移除它，如果不提供INDEX则默认返回队列头部的元素(index=0)。INDEX可以是负数，-1表示队列的最尾部元素，-2表示尾部倒数第二个元素。

如果指定索引处元素不存在(即索引越界)或者队列为空，则返回undef。

记住，peek不会从队列中将数据移除掉，所以操作一个peek出来的数据可能会影响到队列中的对应元素。见后文示例。

insert(INDEX, LIST)

将list中的元素插入到INDEX(从队列头部计数)位置处。负数索引表示从队列尾部开始计数。如果指定的INDEX大于当前队列最大索引值，则直接插入到队列的尾部(正数INDE)或队列的头部(负数INDEX)。例如：
$q->enqueue(1,2,3,4); $q->insert(1, qw(foo bar)); # 1, foo, bar, 2, 3, 4
$q->enqueu(1, 2, 3, 4); $q->insert(-2, qw(foo bar)); # 1, 2, foo, bar, 3, 4
extract()
extract(INDEX)
extract(INDEX, COUNT)
以非阻塞的方式移除并返回指定索引位置处(从队列头部开始计数)指定个数的元素(默认为1个)。如果不给任何参数，则等价于dequeue_nb
。如果元素个数不足，则能返回多少算多少。如果索引位置越界，则返回undef或空列表。

例如：
$q->enqueue(1,2,3,4); my $item = $q->extract(2); # 3，队列包含：1 2 4 my @items =
$q->extract(1, 3); # 返回(2, 4)，队列包含：1
由于peek方法可以直接从队列中获取元素但却不将其从队列中移除，这可能会导致队列中的引用类型元素在队列外被修改。例如，下面的代码：
#!/usr/bin/perl use strict; use warnings; use threads; use threads::shared;
use Thread::Queue; my $q = Thread::Queue->new(); # 放非引用数据到队列 $q->enqueue(1, 2,
3) # 放引用数据到队列 $q->enqueue(['a', 'b', 'c']) my $thr = threads->new( sub { sleep
1; while(my $item = $q->dequeue()){ # 如果是引用，要小心数据被其它线程修改 if (ref $item){
foreach (@$item){ print "ele: $_\n"; } } else { print "item: $item\n"; } } } );
my $num = $q->peek(); $num = 11; # 不影响，因为是非引用数据 my $list = $q->peek(3);
$$list[1] = 'bb'; # 将影响队列中对应元素 $q->end(); # 关闭队列 $thr->join;
为了防止队列中的元素值被其它线程修改，可以使用threads::shared中的lock()将队列锁在一个代码块内。
{ lock($q); my $item = $q->peek(); if($item...){ ... } } # 释放锁

热门工具换一换