问用pthread实现PHP中的多线程文件处理

EN

<?php

// Libraries for reading files
require_once "spooler.php";

// Configuration section ///////////////////////////////////////////////////////

$config["data"] = "data";
$config["threads"] = 20;
$config["timer"] = 1;

// Array to store currently processed files
$config["processed_files"] = array();

// Processing section //////////////////////////////////////////////////////////

$timer = 0;
$pool = new Pool($config["threads"], \ProcessingWorker::class);

while (true) {

    // Read a number of files from the data folder according to the number of thread
    $files = Spooler::read_spool_file($config["data"], $config["threads"]);
    foreach ($files as $file) {
        // Check if the file is already processed
        if (in_array($file, $config["processed_files"])) continue;
        // Submit the file to the worker
        echo "Submitting $file\n";
        $config["processed_files"][$file] = $file;
        $pool->submit(new ProcessingJob($config, $file));
    }

    sleep($config["timer"]);
    $timer++;
}

$pool->shutdown();

// Processing thread section ///////////////////////////////////////////////////

class ProcessingJob extends Stackable {

    private $config;
    private $file;

    public function __construct($config, $file)
    {
        $this->config = $config;
        $this->file = $file;
        $this->complete = false;
    }

    public function run()
    {
        echo "Processing $this->file\n";
        // Pretend we're doing something that takes time
        sleep(mt_rand(1, 10));
        file_put_contents("_LOG", $this->file."\n", FILE_APPEND);

        // Delete the file
        @unlink($this->file);
        // Remove the file from the currently processing list
        unset($this->config["processed_files"][$this->file]);
    }

}

class ProcessingWorker extends Worker {
    public function run() {}
}