安装Pthreads 基本上需要重新编译PHP，加上 --enable-maintainer-zts 参数，但是用这个文档很少；bug会很多很有很多意想不到的问题，生成环境上只能呵呵了，所以这个东西玩玩就算了，真正多线程还是用Python、C等等
一、安装
这里使用的是 php-7.0.2
./configure \ --prefix=https://www.jb51.net/usr/local/php7 \ --with-config-file-path=https://www.jb51.net/etc \ --with-config-file-scan-dir=https://www.jb51.net/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=https://www.jb51.net/usr/local/mysql \ --with-mysqli=https://www.jb51.net/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(); #2 class 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.
<?php class 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); ?>
多线程与共享内存
在共享内存的例子中，没有使用任何锁，仍然可能正常工作，可能工作内存操作本身具备锁的功能
<?php $tmp = tempnam(__FILE__, 'PHP'); $key = ftok($tmp, 'a'); $shmid = shm_attach($key); $counter = 0; shm_put_var( $shmid, 1, $counter ); class CounterThread extends Thread { public function __construct($shmid){ $this->shmid = $shmid; } public function run() { $counter = shm_get_var( $this->shmid, 1 ); $counter++; shm_put_var( $this->shmid, 1, $counter ); printf("Thread #%lu says: %s\n", $this->getThreadId(),$counter); } } for ($i=0;$i<100;$i++){ $threads[] = new CounterThread($shmid); } for ($i=0;$i<100;$i++){ $threads[$i]->start(); } for ($i=0;$i<100;$i++){ $threads[$i]->join(); } shm_remove( $shmid ); shm_detach( $shmid ); ?>
五、 线程同步
有些场景我们不希望 thread->start() 就开始运行程序，而是希望线程等待我们的命令。thread−>wait();测作用是thread−>start()后线程并不会立即运行，只有收到 thread->notify(); 发出的信号后才运行
<?php $tmp = tempnam(__FILE__, 'PHP'); $key = ftok($tmp, 'a'); $shmid = shm_attach($key); $counter = 0; shm_put_var( $shmid, 1, $counter ); class CounterThread extends Thread { public function __construct($shmid){ $this->shmid = $shmid; } public function run() { $this->synchronized(function($thread){ $thread->wait(); }, $this); $counter = shm_get_var( $this->shmid, 1 ); $counter++; shm_put_var( $this->shmid, 1, $counter ); printf("Thread #%lu says: %s\n", $this->getThreadId(),$counter); } } for ($i=0;$i<100;$i++){ $threads[] = new CounterThread($shmid); } for ($i=0;$i<100;$i++){ $threads[$i]->start(); } for ($i=0;$i<100;$i++){ $threads[$i]->synchronized(function($thread){ $thread->notify(); }, $threads[$i]); } for ($i=0;$i<100;$i++){ $threads[$i]->join(); } shm_remove( $shmid ); shm_detach( $shmid ); ?>
六、线程池
一个Pool类