PHP/Beanstalkd : Spawning Multiple workers in parallel

Question

我有一个脚本可以检查哪个 MX 记录属于一个电子邮件地址。我有大约 30 万封电子邮件要检查。因此，单线程进程将花费很长时间。

我有一个带有队列的 beanstalkd，php 正在通过一个文件向它发送电子邮件。但是，我只让一名 worker 执行队列。我目前无法为一个流程生成 10 多个 worker 。

我运行 do_job_mx.php，然后打开一个只包含电子邮件的文件并将它们传递到队列。

从文件中获取电子邮件并放入队列的 php 代码 - do_job_mx.php:

require_once('pheanstalk_init.php');

$pheanstalk = new Pheanstalk_Pheanstalk('127.0.0.1:11300');

$filename = '_blank.txt';
$filename = dirname(__FILE__) . '/in/' . $filename;

foreach (file($filename, FILE_SKIP_EMPTY_LINES) as $line) 
{
    $json = json_encode(array("email" => trim($line)));

    $pheanstalk
        ->useTube('process_mx')
        ->put($json);
}

worker 的 php 代码 - do_worker_process_mx.php:

class Worker 
{
    public function __construct() 
    {
        $this->log('worker process - starting');

        require_once('pheanstalk_init.php');
        $this->pheanstalk = new Pheanstalk_Pheanstalk('127.0.0.1:11300');
    }

    public function __destruct() 
    {
        $this->log('worker process - ending');
    }

    public function run() 
    {
        $this->log('worker process - starting to run');

        while(1) 
        {
            $job = $this->pheanstalk
                ->watch('process_mx')
                ->ignore('default')
                ->reserve();

            $data = json_decode($job->getData(), true);

            $this->process_mx($data);

            $this->pheanstalk->delete($job);
        }
    }

    private function process_mx($data)
    {
        $domain = explode("@", $data['email']);

        dns_get_mx($domain[1], $mx_records);

        $mx_array = explode(".", strtolower($mx_records[0]));

        $mx = array_slice($mx_array, -2, count($mx_array));

        $mx_domain = implode(".", $mx);

        echo $data['email'] . "\n";

        $this->write_file($mx_domain, $data['email']);
    }       

    private function write_file($mx, $email)
    {
        $filename = fopen(dirname(__FILE__) . "/out/" . $mx . ".txt", 'ab+');

        fwrite($filename, $email . "\n");

        fclose($filename);
    }

    private function log($txt) 
    {
        echo $txt . "\n";
    }
}

$worker = new Worker();
$worker->run();

主管 session :

[program:do_worker_process]
command=/usr/bin/php /srv/www/mydev/public_html/esp/do_worker_process_mx.php
numprocs=10
numprocs_start=10
autostart=true
autorestart=true
stopsignal=QUIT
log_stdout=true
logfile=/var/log/supervisor/worker_process_mx.log

我目前无法为一个流程生成 10 多个 worker 。

正在运行的进程数:

# supervisorctl status

do_worker_process RUNNING    pid 44343, uptime 1:46:11

Answer 1

Centos 6 附带:

beanstalkd 1.4.6 主管 2.1.8

我只需要升级到 supervisor 3.0。

现在我有了多人设施。