Thread::Queue()中 dequeue多个元素时的问题

scutdk

我开启一个线程（生产者）从文件中读取数据，并将内容enqueue()到队列中。

同时开始N个线程对线程队列中的数据进行任务处理，现在问题是我希望一次dequeue()多条记录，比如dequeue(100)，但是在这种情况下，如果队列中的元素不足100个的话，就会造成阻塞，甚至在最后队列将要处理完毕只剩几个元素的时候，就会一直阻塞在那里了。但是又不能使用dequeue_nb()，因为有可能生产者还没有enqueue足够的数据进去。怎么样对这种情况进行更好的处理呢？

代码如下：

1. #!/usr/bin/perl
   
2. use strict;
   
3. use threads 1.82;
   
4. use threads::shared;
   
5. use Thread::Queue;
   
6. 
   
7. my $queue = Thread::Queue->new();
   
8. $queue->enqueue(1..9999);
   
9. 
   
10. my @threads = ();
    
11. foreach (1,2,3,4,5) {
    
12.     push @threads,threads->create("worker");
    
13. }
    
14. 
    
15. foreach (1,2,3,4,5) {
    
16.     $queue->enqueue(undef);
    
17. }
    
18. 
    
19. while (1) {
    
20.     foreach (@threads) {
    
21.         my $num = threads->list();
    
22.         print "threads total: $num\n";
    
23.         if ($_->is_running()) {
    
24.             sleep 1;
    
25.             next;
    
26.         }  
    
27. 
    
28.         if ($_->is_joinable()) {
    
29.             $_->join();
    
30.         }  
    
31.    
    
32.         sleep 1;
    
33.     }  
    
34. }
    
35. 
    
36. 
    
37. sub worker {
    
38.     while (my @DataElement = $queue->dequeue(100)) {
    
39.         my $tid = threads->tid();
    
40.         print "Threads ID:$tid\t@DataElement\n";
    
41.     }
    
42. }

复制代码

royalzhang

这段是做什么用的？
foreach (1,2,3,4,5) {
    $queue->enqueue(undef);
}

scutdk

回复 2# royalzhang


    刚开始我是每次dequeue()一个元素的。你提到的那段，作用是结束工作线程，因为工作线程每次dequeue()一个元素，遇到undef就退出while循环并结束整个线程的工作了。

    现在我每次dequeue(100)，你提到的那段就没什么用处了。

royalzhang

回复 3# scutdk


   哦，是的。当每次dequeue一个的时候工作的很好，dequeue多个的时候不行。不知道此模块内部是怎么判断的。填进去多个undef也不行。