Это старая версия документа!

, ,

Threading for PHP - Share Nothing

Примеры

$worker = new Worker();
$task = new class extends Threaded {
    public static $a;
};
$task::$a = new class extends Threaded {};
$worker->start();
$worker->stack($task);
$worker->shutdown();
$task = new class extends Threaded {
    public static $a;
};
$task::$a = new class extends Threaded {};
$pool = new Pool(2);
$pool->submit($task);
$pool->shutdown();
<?php
 
function demanding(...$params) {
  /* you have parameters here */
  return array(rand(), rand());
}
 
class Task extends Collectable {
  public function __construct(Threaded $result, $params) {
    $this->result = $result;
    $this->params = $params;
  }
 
  public function run() {
    $this->result[] = 
      demanding(...$this->params);
  }
 
  protected $result;
  protected $params;
}
 
$pool = new Pool(16);
 
$result = new Threaded();
 
while (@$i++<16) {
  $pool->submit(
    new Task($result, $argv));
}
 
$pool->shutdown();
 
var_dump($result);
?>
class Scanner {
    public function performScan() {
        // Add initial task
        $initialTask = "Task 1";
        TaskQueue::addTask($initialTask);
 
        $i = 0;
        while(true) {
            // Get task from queue
           $task = TaskQueue::getTask();
           if ($task == null)
               break;
 
            // Handle task
            $parser[$i] = new Parser($task);
            $parser[$i]->start();
 
            // Wait
            $thread = $parser[$i];
            $thread->synchronized(function() use($thread) {
                while (!$thread->awake) {
                    $thread->wait();
                }
            });
 
            // Join
            $i++;
        }
 
        // Done
        echo "Done\n";
    }
}
 
class Parser extends Thread {
    private $task;
 
    public function __construct($task) {
        $this->task = $task;
    }
 
    public function run() {
        // Perform a time-consuming operation
        // This operation adds an unknown number of extra tasks
        sleep(1);
 
        // Add new tasks to queue
        foreach(range(0, 4) as $i) {
            TaskQueue::addTask("Task {$i}");
        }
 
        // Notify
        $this->synchronized(function(){
            $this->awake = true;
            $this->notify();
        });
    }
}
 
class TaskQueue {
    private static $queue = array();
 
    public static function addTask($task) {
        self::$queue[] = $task;
        echo "Add task to queue!\n";
    }
 
    public static function getTask() {
        if (sizeof(self::$queue) > 0) {
            $task = array_shift(self::$queue);
            echo "Get task from queue!\n";
            return $task;
        }
    }
}
 
$scanner = new Scanner();
$scanner->performScan();

Документация

Статьи