Введение

Потоковые объекты формируют основу способности pthreads выполнять пользовательский код параллельно; они предоставляют методы синхронизации и различные полезные интерфейсы.

Потоковые объекты, что наиболее важно, обеспечивают неявную безопасность для программиста; все операции с областью объекта безопасны.

Обзор классов

class Threaded implements Collectable , Traversable , Countable , ArrayAccess {
	/* Методы */
	public array chunk ( int $size , bool $preserve )
	public int count ( void )
	public bool extend ( string $class )
	public Threaded from ( Closure $run [, Closure $construct [, array $args ]] )
	public array getTerminationInfo ( void )
	public bool isRunning ( void )
	public bool isTerminated ( void )
	public bool isWaiting ( void )
	public bool lock ( void )
	public bool merge ( mixed $from [, bool $overwrite ] )
	public bool notify ( void )
	public bool notifyOne ( void )
	public bool pop ( void )
	public void run ( void )
	public mixed shift ( void )
	public mixed synchronized ( Closure $block [, mixed $... ] )
	public bool unlock ( void )
	public bool wait ([ int $timeout ] )
}

Содержание

Введение

Потоки Worker имеют постоянный контекст, поэтому в большинстве случаев их следует использовать поверх классов Threads.

Когда Worker запускается, метод Run будет выполнен, но поток не оставится, пока не будет выполнено одно из следующих условий:

• Worker выходит за пределы области действия (больше не остается ссылок)

• программист вызывает завершение работы

• скрипт завершается

Это означает, что программист может повторно использовать контекст во время выполнения; размещение объектов в стеке Worker приведет к тому, что Worker будет выполнять метод выполнения с накоплением объектов.

Обзор классов

class Worker extends Thread implements Traversable , Countable , ArrayAccess {
	/* Методы */
	public int collect ([ Callable $collector ] )
	public int getStacked ( void )
	public bool isShutdown ( void )
	public bool isWorking ( void )
	public bool shutdown ( void )
	public int stack ( Threaded &$work )
	public int unstack ( void )
	/* Наследуемые методы */
	public void Thread::detach ( void )
	public int Thread::getCreatorId ( void )
	public static Thread Thread::getCurrentThread ( void )
	public static int Thread::getCurrentThreadId ( void )
	public int Thread::getThreadId ( void )
	public static mixed Thread::globally ( void )
	public bool Thread::isJoined ( void )
	public bool Thread::isStarted ( void )
	public bool Thread::join ( void )
	public void Thread::kill ( void )
	public bool Thread::start ([ int $options ] )
}

Существует один способ совместного использования данных между Workers, то есть с помощью Stackable:

<?php
class data extends Stackable{
    //private $name;
    public function __construct($_name) {
        //$this->name = $_name;//если вы установите любую переменную, работники получат переменную, поэтому не устанавливайте никакой переменной
        echo __FILE__.'-'.__LINE__.'<br/>'.chr(10);
    }
    public function run(){
        echo __FILE__.'-'.__LINE__.'<br/>'.chr(10);
    }
}
class readWorker extends Worker {
    public function __construct(&$_data) {
        $this->data = $_data;//
    }
    public function run(){
        while(1){
            if($arr=$this->data->shift())//receiving datas
            {
                echo 'Received data:'.print_r($arr,1).chr(10);
            }else usleep(50000);
        }
    }
}
class writeWorker extends Worker {
    public function __construct(&$_data) {
        $this->data = $_data;//
    }
    public function run(){
        while(1){
            $this->data[] = array(time(),rand());//запись данных
            usleep(rand(50000, 1000000));
        }
    }

}
$data = new data('');
$reader = new readWorker($data);
$writer = new writeWorker($data);
$reader->start();
$writer->start();
?>

Также вы можете использовать $readWorker[] = $some_data; затем используйте $this->shift() в readWorker, чтобы поделиться данными с readWorker, но если вы это сделаете, у вас не будет переменных в readWorker, так как все переменные будут сдвинуты на shift();

Содержание

Введение

Представляет объект для сбора мусора.

Обзор интерфейсов

interface Collectable {
	/* Методы */
	public bool isGarbage ( void )
	public void setGarbage ( void )
}

Содержание

Введение

A - это контейнер и контроллер для регулируемого числа Workers.

Объединение в пул обеспечивает более высокий уровень абстракции рабочих функций, включая управление ссылками способом, требуемым pthreads.

Обзор классов

class Pool {
	/* Свойства */
	protected $size ;
	protected $class ;
	protected $workers ;
	protected $ctor ;
	protected $last ;
	/* Методы */
	public int collect ([ Callable $collector ] )
	public Pool __construct ( int $size [, string $class [, array $ctor ]] )
	public void resize ( int $size )
	public void shutdown ( void )
	public int submit ( Threaded $task )
	public int submitTo ( int $worker , Threaded $task )
}

Простой пример с Collectable (в основном Thread предназначен для Pool) и Pool

<?php

class job extends Collectable {
  public $val;

  public function __construct($val){
    // инициировать некоторые свойства
    $this->val = $val;
  }
  public function run(){
    // поработай немного
    $this->val = $this->val . file_get_contents('http://www.example.com/', null, null, 3, 20);
    $this->setGarbage();
  }
}

// Максимум 3 темы будут работать одновременно
$p = new Pool(3);

$tasks = array(
  new job('0'),
  new job('1'),
  new job('2'),
  new job('3'),
  new job('4'),
  new job('5'),
  new job('6'),
  new job('7'),
  new job('8'),
  new job('9'),
  new job('10'),
);
// Добавить задачи в очередь пула
foreach ($tasks as $task) {
  $p->submit($task);
}

// завершение работы будет ожидать завершения текущей очереди
$p->shutdown();
// проверка / чтение результатов сборки мусора
$p->collect(function($checkingTask){
  echo $checkingTask->val;
  return $checkingTask->isGarbage();
});

?>

Обратите внимание, что при использовании функции сбора важно расширить класс пула, чтобы вы могли продолжать проверять готовые потоки, пока все они не будут завершены.

<?php
class TestWork extends Threaded {
    protected $complete;
    //$pData-это данные, отправляемые в рабочий поток для выполнения его работы.
    public function __construct($pData){
        //перенести все переменные в локальные переменные
        $this->complete = false;
        $this->testData = $pData;
    }
    //Здесь будет сделана вся ваша работа.
    public function run(){
        usleep(2000000); //сон 2 секунды для того чтобы сымитировать большую работу
        $this->complete = true;
    }
    public function isGarbage() {
        return $this->complete;
    }
}
class ExamplePool extends Pool
{
    public $data = array();
    public function process()
    {
        // Выполните этот цикл,
        // пока у нас есть задания в пуле
        while (count($this->work)) {
            $this->collect(function (TestWork $task) {
                // Если задача была помечена как выполненная,
                // соберите ее результаты
                if ($task->isGarbage()) {
                    $tmpObj = new stdclass();
                    $tmpObj->complete = $task->complete;
                    // это то, как вы возвращаете свои законченные данные [доступно для $pool->process()]
                    $this->data[] = $tmpObj;
                }
                return $task->isGarbage();
            });
        }
        // Все работы выполнены,
        // мы можем закрыть pool
        $this->shutdown();
        return $this->data;
    }
}
$pool = new ExamplePool(3);
$testData = 'asdf';
for($i=0;$i<5;$i++) {
    $pool->submit(new TestWork($testData));
}
$retArr = $pool->process(); //получить все результаты
echo '<pre>';
print_r($retArr); //вернуть массив результатов (и, возможно, ошибки)
echo '</pre>';
?>

Пример класса, чтобы продемонстрировать использование пула / рабочего механизма, а также показать несколько трюков и подсказок ;)

<?php
class Config extends Threaded{    // общий глобальный объект
    protected $val=0, $val2=0;
    protected function inc(){++$this->val;}    // защищенная синхронизация на объект
    public function inc2(){++$this->val2;}    // нет синхронизации
}
class WorkerClass extends Worker{
    protected static $worker_id_next = -1;
    protected $worker_id;
    protected $config;
    public function __construct($config){
        $this->worker_id = ++static::$worker_id_next;    // статические члены не доступны в потоке, но находятся в "основном потоке"
        $this->config = $config;
    }
    public function run(){
        global $config;
        $config = $this->config;    // ПРИМЕЧАНИЕ: настройка по ссылке НЕ РАБОТАЕТ
        global $worker_id;
        $worker_id = $this->worker_id;
        echo "working context {$worker_id} is created!\n";
        //$this->say_config();    // глобально синхронизированная функция.
    }
    protected function say_config(){    // 'protected' синхронизируется побочным объектом, поэтому НЕ БУДЕТ работать между несколькими экземплярами,
        global $config;        // вы можете использовать общий объект $config в качестве источника синхронизации.
        $config->synchronized(function() use (&$config){    // ПРИМЕЧАНИЕ: вы можете использовать здесь Closures, 
															// но если вы прикрепите Closure к объекту с потоком, он будет уничтожен, поскольку не может быть сериализован
            var_dump($config);
        });
    }
}
class Task extends Stackable{    // Стекируемый до сих пор существует, просто как-то исчез из документации (вероятно, по ошибке). 
								// Для получения более подробной информации смотрите документацию к старой версии.
    protected $set;
    public function __construct($set){
        $this->set = $set;
    }
    public function run(){
        global $worker_id;
        echo "task is running in {$worker_id}!\n";
        usleep(mt_rand(1,100)*100);
        $config = $this->getConfig();
        $val = $config->arr->shift();
        $config->arr[] = $this->set;
        for ($i = 0 ; $i < 1000; ++$i){
            $config->inc();
            $config->inc2();
        }
    }
    public function getConfig(){
        global $config;    // WorkerClass устанавливает это в области видимости потока, может быть повторно использовано Задачами для дополнительного асинхронного источника данных. 
							//(то есть: пул соединений или очередь задач к демультиплексору)
        return $config;
    }
}

$config = new Config;
$config->arr = new \Threaded();
$config->arr->merge(array(1,2,3,4,5,6));
class PoolClass extends Pool{
    public function worker_list(){
        if ($this->workers !== null)
            return array_keys($this->workers);
        return null;
    }
}
$pool = new PoolClass(3, 'WorkerClass', [$config] );
$pool->worker_list();
//$pool->submitTo(0,new Task(-10));    // submitTo не пытается создать worker

$spammed_id = -1;
for ($i = 1; $i <= 100; ++$i){    // добавить некоторые задания
    if ($spammed_id == -1 && ($x = $pool->worker_list())!= null && @$x[2]){
        $spammed_id = $x[2];
        echo "spamming worker {$spammed_id} with lots of tasks from now on\n";
    }
    if ($spammed_id != -1 && ($i % 5) == 0)    // каждая пятая работа направляется одному работнику, поэтому она имеет 20% от общего числа рабочих мест 
												//(с 3 работниками она должна делать ~33%, а не (33+20)%, поэтому делегируйте только работнику, если вы планируете также балансировать... )
        $pool->submitTo($spammed_id,new Task(10*$i));   
    else
        $pool->submit(new Task(10*$i));
}
$pool->shutdown();
var_dump($config); // "val" составляет ровно 100000, "val2", наверное, немного меньше
// также: если вы отключите спамер, вы увидите, что порядок "arr" является случайным.
?>

Содержание

Введение

Кроссплатформенная, встроенная реализация именованных и неназванных счетных объектов мьютекса.

Мьютекс - это объект взаимного исключения, который ограничивает доступ к общему ресурсу (например, файлу) одним экземпляром. Счетные мьютексы получают мьютекс один раз и внутренне отслеживают количество блокировок мьютекса. Мьютекс разблокируется, как только он выходит за пределы области действия, или разблокируется столько же раз, сколько он был заблокирован.

Обзор классов

class SyncMutex {
	/* Методы */
	public __construct ([ string $name ] )
	public bool lock ([ int $wait = -1 ] )
	public bool unlock ([ bool $all = FALSE ] )
}

Содержание

Введение

Кроссплатформенная, встроенная реализация именованных и неназванных семафорных объектов.

Семафор ограничивает доступ к ограниченному ресурсу ограниченным числом экземпляров. Семафоры отличаются от мьютексов тем, что они позволяют нескольким экземплярам одновременно обращаться к ресурсу, в то время как мьютекс разрешает только один экземпляр одновременно.

Обзор классов

class SyncSemaphore {
	/* Методы */
	public __construct ([ string $name [, int $initialval = 1 [, bool $autounlock = TRUE ]]] )
	public bool lock ([ int $wait = -1 ] )
	public bool unlock ([ int &$prevcount ] )
}

Содержание

Введение

Кроссплатформенная встроенная реализация именованных и неназванных объектов событий. Поддерживаются как автоматические, так и ручные объекты событий.

Объект события ожидает, без опроса, объект, который будет запущен/установлен. Один экземпляр ожидает объект события, в то время как другой экземпляр запускает/задает событие. Объекты событий полезны везде, где длительный процесс в противном случае опрашивает ресурс (например, проверяет, нужно ли обрабатывать загруженные данные).

Обзор классов

class SyncEvent {
	/* Методы */
	public __construct ([ string $name [, bool $manual = FALSE [, bool $prefire = FALSE ]]] )
	public bool fire ( void )
	public bool reset ( void )
	public bool wait ([ int $wait = -1 ] )
}

Введение

Кроссплатформенная встроенная реализация именованных и неназванных объектов чтения и записи.

Объект reader-writer позволяет многим читателям или одному писателю получить доступ к ресурсу. Это эффективное решение для управления ресурсами, где доступ будет только для чтения, но запись иногда требуется.

Обзор классов

class  SyncReaderWriter {
	/* Методы */
	public __construct ([ string $name [, bool $autounlock = TRUE ]] )
	public bool readlock ([ int $wait = -1 ] )
	public bool readunlock ( void )
	public bool writelock ([ int $wait = -1 ] )
	public bool writeunlock ( void )
}

Содержание