php pthreads多线程的安装与使用

./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

<?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();}?>

<?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);?>

<?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 );?>

<?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 );?> 

<?phpclass Update extends Thread {public $running = false;public $row = array();public function __construct($row) {$this->row = $row;$this->sql = null;}public function run() {if(strlen($this->row['bankno']) > 100 ){$bankno = safenet_decrypt($this->row['bankno']);}else{$error = sprintf("%s, %s/r/n",$this->row['id'], $this->row['bankno']);file_put_contents("bankno_error.log", $error, FILE_APPEND);}if( strlen($bankno) > 7 ){$sql = sprintf("update members set bankno = '%s' where id = '%s';", $bankno, $this->row['id']);$this->sql = $sql;}printf("%s/n",$this->sql);}}class Pool {public $pool = array();public function __construct($count) {$this->count = $count;}public function push($row){if(count($this->pool) < $this->count){$this->pool[] = new Update($row);return true;}else{return false;}}public function start(){foreach ( $this->pool as $id => $worker){$this->pool[$id]->start();}}public function join(){foreach ( $this->pool as $id => $worker){$this->pool[$id]->join();}}public function clean(){foreach ( $this->pool as $id => $worker){if(! $worker->isRunning()){unset($this->pool[$id]);}}}}try {$dbh = new PDO("mysql:host=" . str_replace(':', ';port=', $dbhost) . ";dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF8/'',PDO::MYSQL_ATTR_COMPRESS => true));$sql = "select id,bankno from members order by id desc";$row = $dbh->query($sql);$pool = new Pool(5);while($member = $row->fetch(PDO::FETCH_ASSOC)){while(true){if($pool->push($member)){ //压入任务到池中break;}else{ //如果池已经满，就开始启动线程$pool->start();$pool->join();$pool->clean();}}}$pool->start();$pool->join();$dbh = null;} catch (Exception $e) {echo '[' , date('H:i:s') , ']', '系统错误', $e->getMessage(), "/n";}?>

<?phpclass Update extends Thread {public $running = false;public $row = array();public function __construct($row) {$this->row = $row;$this->sql = null;//print_r($this->row);}public function run() {if(strlen($this->row['bankno']) > 100 ){$bankno = safenet_decrypt($this->row['bankno']);}else{$error = sprintf("%s, %s/r/n",$this->row['id'], $this->row['bankno']);file_put_contents("bankno_error.log", $error, FILE_APPEND);}if( strlen($bankno) > 7 ){$sql = sprintf("update members set bankno = '%s' where id = '%s';", $bankno, $this->row['id']);$this->sql = $sql;}printf("%s/n",$this->sql);}}try {$dbh = new PDO("mysql:host=" . str_replace(':', ';port=', $dbhost) . ";dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF8/'',PDO::MYSQL_ATTR_COMPRESS => true));$sql = "select id,bankno from members order by id desc limit 50";$row = $dbh->query($sql);$pool = array();while($member = $row->fetch(PDO::FETCH_ASSOC)){$id = $member['id'];while (true){if(count($pool) < 5){$pool[$id] = new Update($member);$pool[$id]->start();break;}else{foreach ( $pool as $name => $worker){if(! $worker->isRunning()){unset($pool[$name]);}}}}}$dbh = null;} catch (Exception $e) {echo '【' , date('H:i:s') , '】', '【系统错误】', $e->getMessage(), "/n";}?>

<?phpclass WebWorker extends Worker {public function __construct(SafeLog $logger) {$this->logger = $logger;}protected $loger;}class WebWork extends Stackable {public function isComplete() {return $this->complete;}public function run() {$this->worker->logger->log("%s executing in Thread #%lu",__CLASS__, $this->worker->getThreadId());$this->complete = true;}protected $complete;}class SafeLog extends Stackable {protected function log($message, $args = []) {$args = func_get_args();if (($message = array_shift($args))) {echo vsprintf("{$message}/n", $args);}}}$pool = new Pool(8, /WebWorker::class, [new SafeLog()]);$pool->submit($w=new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->submit(new WebWork());$pool->shutdown();$pool->collect(function($work){return $work->isComplete();});var_dump($pool); 

<?php$fp = fopen("/tmp/lock.txt", "r+");if (flock($fp, LOCK_EX)) { // 进行排它型锁定ftruncate($fp, 0); // truncate filefwrite($fp, "Write something here/n");fflush($fp); // flush output before releasing the lockflock($fp, LOCK_UN); // 释放锁定} else {echo "Couldn't get the lock!";}fclose($fp);$fp = fopen('/tmp/lock.txt', 'r+');if(!flock($fp, LOCK_EX | LOCK_NB)) {echo 'Unable to obtain lock';exit(-1);}fclose($fp);?>

<?phpclass Work extends Stackable {public function __construct() {}public function run() {$dbh = $this->worker->getConnection();$sql = "select id,name from members order by id desc limit ";$row = $dbh->query($sql);while($member = $row->fetch(PDO::FETCH_ASSOC)){print_r($member);}}}class ExampleWorker extends Worker {public static $dbh;public function __construct($name) {}/** The run method should just prepare the environment for the work that is coming ...*/public function run(){self::$dbh = new PDO('mysql:host=...;dbname=example','www','');}public function getConnection(){return self::$dbh;}}$worker = new ExampleWorker("My Worker Thread");$work=new Work();$worker->stack($work);$worker->start();$worker->shutdown();?> 

# cat pool.php<?phpclass ExampleWorker extends Worker {public function __construct(Logging $logger) {$this->logger = $logger;}protected $logger;}/* the collectable class implements machinery for Pool::collect */class Work extends Stackable {public function __construct($number) {$this->number = $number;}public function run() {$dbhost = 'db.example.com'; // 数据库服务器$dbuser = 'example.com'; // 数据库用户名$dbpw = 'password'; // 数据库密码$dbname = 'example_real';$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF/'',PDO::MYSQL_ATTR_COMPRESS => true,PDO::ATTR_PERSISTENT => true));$sql = "select OPEN_TIME, `COMMENT` from MT_TRADES where LOGIN='".$this->number['name']."' and CMD='' and `COMMENT` = '".$this->number['order'].":DEPOSIT'";#echo $sql;$row = $dbh->query($sql);$mt_trades = $row->fetch(PDO::FETCH_ASSOC);if($mt_trades){$row = null;$sql = "UPDATE db_example.accounts SET paystatus='成功', deposit_time='".$mt_trades['OPEN_TIME']."' where `order` = '".$this->number['order']."';";$dbh->query($sql);#printf("%s/n",$sql);}$dbh = null;printf("runtime: %s, %s, %s/n", date('Y-m-d H:i:s'), $this->worker->getThreadId() ,$this->number['order']);}}class Logging extends Stackable {protected static $dbh;public function __construct() {$dbhost = 'db.example.com'; // 数据库服务器$dbuser = 'example.com'; // 数据库用户名$dbpw = 'password'; // 数据库密码$dbname = 'example_real'; // 数据库名self::$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF/'',PDO::MYSQL_ATTR_COMPRESS => true));}protected function log($message, $args = []) {$args = func_get_args();if (($message = array_shift($args))) {echo vsprintf("{$message}/n", $args);}}protected function getConnection(){return self::$dbh;}}$pool = new Pool(, /ExampleWorker::class, [new Logging()]);$dbhost = 'db.example.com'; // 数据库服务器$dbuser = 'example.com'; // 数据库用户名$dbpw = 'password'; // 数据库密码$dbname = 'db_example';$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF/'',PDO::MYSQL_ATTR_COMPRESS => true));$sql = "select `order`,name from accounts where deposit_time is null order by id desc";$row = $dbh->query($sql);while($account = $row->fetch(PDO::FETCH_ASSOC)){$pool->submit(new Work($account));}$pool->shutdown();?>　

<?phpclass ExampleWorker extends Worker {#public function __construct(Logging $logger) {# $this->logger = $logger;#}#protected $logger;protected static $dbh;public function __construct() {}public function run(){$dbhost = 'db.example.com'; // 数据库服务器$dbuser = 'example.com'; // 数据库用户名$dbpw = 'password'; // 数据库密码$dbname = 'example'; // 数据库名self::$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF/'',PDO::MYSQL_ATTR_COMPRESS => true,PDO::ATTR_PERSISTENT => true));}protected function getInstance(){return self::$dbh;}}/* the collectable class implements machinery for Pool::collect */class Work extends Stackable {public function __construct($data) {$this->data = $data;#print_r($data);}public function run() {#$this->worker->logger->log("%s executing in Thread #%lu", __CLASS__, $this->worker->getThreadId() );try {$dbh = $this->worker->getInstance();#print_r($dbh);$id = $this->data['id'];$mobile = safenet_decrypt($this->data['mobile']);#printf("%d, %s /n", $id, $mobile);if(strlen($mobile) > ){$mobile = substr($mobile, -);}if($mobile == 'null'){# $sql = "UPDATE members_digest SET mobile = '".$mobile."' where id = '".$id."'";# printf("%s/n",$sql);# $dbh->query($sql);$mobile = '';$sql = "UPDATE members_digest SET mobile = :mobile where id = :id";}else{$sql = "UPDATE members_digest SET mobile = md(:mobile) where id = :id";}$sth = $dbh->prepare($sql);$sth->bindValue(':mobile', $mobile);$sth->bindValue(':id', $id);$sth->execute();#echo $sth->debugDumpParams();}catch(PDOException $e) {$error = sprintf("%s,%s/n", $mobile, $id );file_put_contents("mobile_error.log", $error, FILE_APPEND);}#$dbh = null;printf("runtime: %s, %s, %s, %s/n", date('Y-m-d H:i:s'), $this->worker->getThreadId() ,$mobile, $id);#printf("runtime: %s, %s/n", date('Y-m-d H:i:s'), $this->number);}}$pool = new Pool(, /ExampleWorker::class, []);#foreach (range(, ) as $number) {# $pool->submit(new Work($number));#}$dbhost = 'db.example.com'; // 数据库服务器$dbuser = 'example.com'; // 数据库用户名$dbpw = 'password'; // 数据库密码$dbname = 'example';$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array(PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES /'UTF/'',PDO::MYSQL_ATTR_COMPRESS => true));#print_r($dbh);#$sql = "select id, mobile from members where id < :id";#$sth = $dbh->prepare($sql);#$sth->bindValue(':id',);#$sth->execute();#$result = $sth->fetchAll();#print_r($result);##$sql = "UPDATE members_digest SET mobile = :mobile where id = :id";#$sth = $dbh->prepare($sql);#$sth->bindValue(':mobile', 'aa');#$sth->bindValue(':id','');#echo $sth->execute();#echo $sth->queryString;#echo $sth->debugDumpParams();$sql = "select id, mobile from members order by id asc"; // limit ";$row = $dbh->query($sql);while($members = $row->fetch(PDO::FETCH_ASSOC)){#$order = $account['order'];#printf("%s/n",$order);//print_r($members);$pool->submit(new Work($members));#unset($account['order']);}$pool->shutdown();?> 

<?php$dbh = new PDO('mysql:host=localhost;dbname=test', $user, $pass, array(PDO::ATTR_PERSISTENT => true));?>