multithreading - Perl 队列和线程

Question

我正在尝试完成以下工作:

有一个从一个非常大的文件中读取数据的线程说
10GB 并将它们插入队列。 (我不希望队列
变得非常大)

而buildQueue线程正在将数据推送到队列的同时有
大约 5 个工作线程出队并处理数据。

我已经尝试过，但由于 buildQueue 中的连续循环，我的其他线程无法访问线。

我的方法可能完全错误。感谢您的帮助，非常感谢。

这是 buildQueue的代码:

sub buildQueue {
    print "Enter a file name: ";
    my $dict_path = <STDIN>;
    chomp($dict_path);
    open DICT_FILE, $dict_path or die("Sorry, could not open file!");
    while (1) {
        if (<DICT_FILE>) {
            if ($queue->pending() < 100) {
                 my $query = <DICT_FILE>;
                 chomp($query);
                 $queue->enqueue($query);
                 my $count = $queue->pending();
                 print "Queue Size: $count Query: $query\n";
            }
        }
    }
}

正如我所期望的，当这个线程被执行时，不会再执行其他任何东西，因为这个线程不会完成。

my $builder = new Thread(&buildQueue);

由于构建器线程将运行很长时间，我永远不会创建工作线程。

这是整个代码:

#!/usr/bin/perl -w
use strict;
use Thread;
use Thread::Queue;


my $queue = new Thread::Queue();
my @threads;

sub buildQueue {
    print "Enter a file name: ";
    my $dict_path = <STDIN>;
    chomp($dict_path);
    open dict_file, $dict_path or die("Sorry, could not open file!");
    while (1) {
        if (<dict_file>) {
            if ($queue->pending() < 100) {
                 my $query = <dict_file>;
                 chomp($query);
                 $queue->enqueue($query);
                 my $count = $queue->pending();
                 print "Queue Size: $count Query: $query\n";
            }
        }
    }
}

sub processor {
    my $query;
    while (1) {
        if ($query = $queue->dequeue) {
            print "$query\n";
        }
    }
}

my $builder = new Thread(&buildQueue);
push @threads, new Thread(&processor) for 1..5;

Answer 1

您需要标记您希望线程何时退出(通过 join or detach )。事实上，你有无限循环，没有 last报表要打出来也是个问题。

编辑:我也忘记了一个很重要的部分! Each worker thread will block, waiting for another item to process off of the queue until they get an undef in the queue .这就是为什么我们专门排队 undef队列构建器完成后，每个线程执行一次。

尝试:

#!/usr/bin/perl -w
use strict;
use threads;
use Thread::Queue;


my $queue = new Thread::Queue();
our @threads; #Do you really need our instead of my?

sub buildQueue
{
    print "Enter a file name: ";
    my $dict_path = <STDIN>;
    chomp($dict_path);

    #Three-argument open, please!
    open my $dict_file, "<",$dict_path or die("Sorry, could not open file!");
    while(my $query=<$dict_file>)
    {
        chomp($query);
        while(1)
        {   #Wait to see if our queue has < 100 items...
            if ($queue->pending() < 100) 
            {
                $queue->enqueue($query);
                print "Queue Size: " . $queue->pending . "\n";
                last; #This breaks out of the infinite loop
            }
        }
    }
    close($dict_file);
    foreach(1..5)
    {
        $queue->enqueue(undef);
    }
}

sub processor 
{
    my $query;
    while ($query = $queue->dequeue) 
    {
        print "Thread " . threads->tid . " got $query\n";
    }
}

my $builder=threads->create(\&buildQueue);
push @threads,threads->create(\&process) for 1..5;

#Waiting for our threads to finish.
$builder->join;
foreach(@threads)
{
    $_->join;
}