Thread::Queue()中 dequeue多個元素時的問題

scutdk

我開啟一個線程（生產(chǎn)者）從文件中讀取數(shù)據(jù)，并將內(nèi)容enqueue()到隊列中。

同時開始N個線程對線程隊列中的數(shù)據(jù)進行任務處理，現(xiàn)在問題是我希望一次dequeue()多條記錄，比如dequeue(100)，但是在這種情況下，如果隊列中的元素不足100個的話，就會造成阻塞，甚至在最后隊列將要處理完畢只剩幾個元素的時候，就會一直阻塞在那里了。但是又不能使用dequeue_nb()，因為有可能生產(chǎn)者還沒有enqueue足夠的數(shù)據(jù)進去。怎么樣對這種情況進行更好的處理呢？

代碼如下：

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循環(huán)并結束整個線程的工作了。

    現(xiàn)在我每次dequeue(100)，你提到的那段就沒什么用處了。

royalzhang

回復 3# scutdk


   哦，是的。當每次dequeue一個的時候工作的很好，dequeue多個的時候不行。不知道此模塊內(nèi)部是怎么判斷的。填進去多個undef也不行。