php pthreads多线程的安装与使用

安装Pthreads 基本上需要重新编译PHP，加上 --enable-maintainer-zts 参数，但是用这个文档很少；bug会很多很有很多意想不到的问题，生成环境上只能呵呵了，所以这个东西玩玩就算了，真正多线程还是用Python、C等等

一、安装

这里使用的是 php-7.0.2

./configure /--prefix=/usr/local/php7 /--with-config-file-path=/etc /--with-config-file-scan-dir=/etc/php.d /--enable-debug /--enable-maintainer-zts /--enable-pcntl /--enable-fpm /--enable-opcache /--enable-embed=shared /--enable-json=shared /--enable-phpdbg /--with-curl=shared /--with-mysql=/usr/local/mysql /--with-mysqli=/usr/local/mysql/bin/mysql_config /--with-pdo-mysql

make && make install

安装pthreads

pecl install pthreads

二、Thread

<?php#1$thread = new class extends Thread {public function run() {echo "Hello World {$this->getThreadId()}/n"; } };$thread->start() && $thread->join();#2class workerThread extends Thread { public function __construct($i){$this->i=$i;}public function run(){while(true){echo $this->i."/n";sleep(1);} } }for($i=0;$i<50;$i++){$workers[$i]=new workerThread($i);$workers[$i]->start();}?>

三、 Worker 与 Stackable

Stackables are tasks that are executed by Worker threads. You can synchronize with, read, and write Stackable objects before, after and during their execution.

<?phpclass SQLQuery extends Stackable {public function __construct($sql) {$this->sql = $sql;}public function run() {$dbh = $this->worker->getConnection();$row = $dbh->query($this->sql);while($member = $row->fetch(PDO::FETCH_ASSOC)){print_r($member);}}}class ExampleWorker extends Worker {public static $dbh;public function __construct($name) {}public function run(){self::$dbh = new PDO('mysql:host=10.0.0.30;dbname=testdb','root','123456');}public function getConnection(){return self::$dbh;}}$worker = new ExampleWorker("My Worker Thread");$sql1 = new SQLQuery('select * from test order by id desc limit 1,5');$worker->stack($sql1);$sql2 = new SQLQuery('select * from test order by id desc limit 5,5');$worker->stack($sql2);$worker->start();$worker->shutdown();?>

四、 互斥锁

什么情况下会用到互斥锁？在你需要控制多个线程同一时刻只能有一个线程工作的情况下可以使用。一个简单的计数器程序，说明有无互斥锁情况下的不同

<?php$counter = 0;$handle=fopen("/tmp/counter.txt", "w");fwrite($handle, $counter );fclose($handle);class CounterThread extends Thread {public function __construct($mutex = null){$this->mutex = $mutex;$this->handle = fopen("/tmp/counter.txt", "w+");}public function __destruct(){fclose($this->handle);}public function run() {if($this->mutex)$locked=Mutex::lock($this->mutex);$counter = intval(fgets($this->handle));$counter++;rewind($this->handle);fputs($this->handle, $counter );printf("Thread #%lu says: %s/n", $this->getThreadId(),$counter);if($this->mutex)Mutex::unlock($this->mutex);}}//没有互斥锁for ($i=0;$i<50;$i++){$threads[$i] = new CounterThread();$threads[$i]->start();}//加入互斥锁$mutex = Mutex::create(true);for ($i=0;$i<50;$i++){$threads[$i] = new CounterThread($mutex);$threads[$i]->start();}Mutex::unlock($mutex);for ($i=0;$i<50;$i++){$threads[$i]->join();}Mutex::destroy($mutex);?>