当前位置：首页 > PHP教程 > php高级应用 > 列表

php pthreads多线程的安装与使用

发布:smiling　来源: PHP粉丝网 　添加日期：2021-07-06 10:35:06　浏览: 评论：0　

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

一、安装

这里使用的是 php-7.0.2

  1. ./configure \ 
  2. --prefix=/usr/local/php7 \ 
  3. --with-config-file-path=/etc \ 
  4. --with-config-file-scan-dir=/etc/php.d \ 
  5. --enable-debug \ 
  6. --enable-maintainer-zts \ 
  7. --enable-pcntl \ 
  8. --enable-fpm \ 
  9. --enable-opcache \ 
  10. --enable-embed=shared \ 
  11. --enable-json=shared \ 
  12. --enable-phpdbg \ 
  13. --with-curl=shared \ 
  14. --with-mysql=/usr/local/mysql \ 
  15. --with-mysqli=/usr/local/mysql/bin/mysql_config \ 
  16. --with-pdo-mysql 
  17. make && make install 

安装pthreads

pecl install pthreads

二、Thread

  1. <?php 
  2. #1 
  3. $thread = new class extends Thread { 
  4. public function run() { 
  5. echo "Hello World {$this->getThreadId()}\n";  
  6. }  
  7. }; 
  8. $thread->start() && $thread->join(); 
  9. #2 
  10. class workerThread extends Thread {  
  11. public function __construct($i){ 
  12. $this->i=$i; 
  13. } 
  14. public function run(){ 
  15. while(true){ 
  16. echo $this->i."\n"; 
  17. sleep(1); 
  18. }  
  19. }  
  20. } 
  21. for($i=0;$i<50;$i++){ 
  22. $workers[$i]=new workerThread($i); 
  23. $workers[$i]->start(); 
  24. } 
  25. ?> 

三、 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.

  1. <?php 
  2. class SQLQuery extends Stackable { 
  3. public function __construct($sql) { 
  4. $this->sql = $sql; 
  5. } 
  6. public function run() { 
  7. $dbh = $this->worker->getConnection(); 
  8. $row = $dbh->query($this->sql); 
  9. while($member = $row->fetch(PDO::FETCH_ASSOC)){ 
  10. print_r($member); 
  11. } 
  12. } 
  13. } 
  14. class ExampleWorker extends Worker { 
  15. public static $dbh; 
  16. public function __construct($name) { 
  17. } 
  18. public function run(){ 
  19. self::$dbh = new PDO('mysql:host=10.0.0.30;dbname=testdb','root','123456'); 
  20. } 
  21. public function getConnection(){ 
  22. return self::$dbh; 
  23. } 
  24. } 
  25. $worker = new ExampleWorker("My Worker Thread"); 
  26. $sql1 = new SQLQuery('select * from test order by id desc limit 1,5'); 
  27. $worker->stack($sql1); 
  28. $sql2 = new SQLQuery('select * from test order by id desc limit 5,5'); 
  29. $worker->stack($sql2); 
  30. $worker->start(); 
  31. $worker->shutdown(); 
  32. ?> 

四、 互斥锁

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

  1. <?php 
  2. $counter = 0; 
  3. $handle=fopen("/tmp/counter.txt", "w"); 
  4. fwrite($handle, $counter ); 
  5. fclose($handle); 
  6. class CounterThread extends Thread { 
  7. public function __construct($mutex = null){ 
  8. $this->mutex = $mutex; 
  9. $this->handle = fopen("/tmp/counter.txt", "w+"); 
  10. } 
  11. public function __destruct(){ 
  12. fclose($this->handle); 
  13. } 
  14. public function run() { 
  15. if($this->mutex) 
  16. $locked=Mutex::lock($this->mutex); 
  17. $counter = intval(fgets($this->handle)); 
  18. $counter++; 
  19. rewind($this->handle); 
  20. fputs($this->handle, $counter ); 
  21. printf("Thread #%lu says: %s\n", $this->getThreadId(),$counter); 
  22. if($this->mutex) 
  23. Mutex::unlock($this->mutex); 
  24. } 
  25. } 
  26. //没有互斥锁 
  27. for ($i=0;$i<50;$i++){ 
  28. $threads[$i] = new CounterThread(); 
  29. $threads[$i]->start(); 
  30. } 
  31. //加入互斥锁 
  32. $mutex = Mutex::create(true); 
  33. for ($i=0;$i<50;$i++){ 
  34. $threads[$i] = new CounterThread($mutex); 
  35. $threads[$i]->start(); 
  36. } 
  37. Mutex::unlock($mutex); 
  38. for ($i=0;$i<50;$i++){ 
  39. $threads[$i]->join(); 
  40. } 
  41. Mutex::destroy($mutex); 
  42. ?> 

多线程与共享内存

在共享内存的例子中，没有使用任何锁，仍然可能正常工作，可能工作内存操作本身具备锁的功能。

  1. <?php 
  2. $tmp = tempnam(__FILE__, 'PHP'); 
  3. $key = ftok($tmp, 'a'); 
  4. $shmid = shm_attach($key); 
  5. $counter = 0; 
  6. shm_put_var( $shmid, 1, $counter ); 
  7. class CounterThread extends Thread { 
  8. public function __construct($shmid){ 
  9. $this->shmid = $shmid; 
  10. } 
  11. public function run() { 
  12. $counter = shm_get_var( $this->shmid, 1 ); 
  13. $counter++; 
  14. shm_put_var( $this->shmid, 1, $counter ); 
  15. printf("Thread #%lu says: %s\n", $this->getThreadId(),$counter); 
  16. } 
  17. } 
  18. for ($i=0;$i<100;$i++){ 
  19. $threads[] = new CounterThread($shmid); 
  20. } 
  21. for ($i=0;$i<100;$i++){ 
  22. $threads[$i]->start(); 
  23. } 
  24. for ($i=0;$i<100;$i++){ 
  25. $threads[$i]->join(); 
  26. } 
  27. shm_remove( $shmid ); 
  28. shm_detach( $shmid ); 
  29. ?> 

五、 线程同步

有些场景我们不希望 thread->start() 就开始运行程序，而是希望线程等待我们的命令。thread−>wait();测作用是thread−>start()后线程并不会立即运行，只有收到 thread->notify(); 发出的信号后才运行

  1. <?php 
  2. $tmp = tempnam(__FILE__, 'PHP'); 
  3. $key = ftok($tmp, 'a'); 
  4. $shmid = shm_attach($key); 
  5. $counter = 0; 
  6. shm_put_var( $shmid, 1, $counter ); 
  7. class CounterThread extends Thread { 
  8. public function __construct($shmid){ 
  9. $this->shmid = $shmid; 
  10. } 
  11. public function run() { 
  12. $this->synchronized(function($thread){ 
  13. $thread->wait(); 
  14. }, $this); 
  15. $counter = shm_get_var( $this->shmid, 1 ); 
  16. $counter++; 
  17. shm_put_var( $this->shmid, 1, $counter ); 
  18. printf("Thread #%lu says: %s\n", $this->getThreadId(),$counter); 
  19. } 
  20. } 
  21. for ($i=0;$i<100;$i++){ 
  22. $threads[] = new CounterThread($shmid); 
  23. } 
  24. for ($i=0;$i<100;$i++){ 
  25. $threads[$i]->start(); 
  26. } 
  27. for ($i=0;$i<100;$i++){ 
  28. $threads[$i]->synchronized(function($thread){ 
  29. $thread->notify(); 
  30. }, $threads[$i]); 
  31. } 
  32. for ($i=0;$i<100;$i++){ 
  33. $threads[$i]->join(); 
  34. } 
  35. shm_remove( $shmid ); 
  36. shm_detach( $shmid ); 
  37. ?> 

六、线程池

一个Pool类

  1. <?php 
  2. class Update extends Thread { 
  3. public $running = false; 
  4. public $row = array(); 
  5. public function __construct($row) { 
  6. $this->row = $row; 
  7. $this->sql = null; 
  8. } 
  9. public function run() { 
  10. if(strlen($this->row['bankno']) > 100 ){ 
  11. $bankno = safenet_decrypt($this->row['bankno']); 
  12. }else{ 
  13. $error = sprintf("%s, %s\r\n",$this->row['id'], $this->row['bankno']); 
  14. file_put_contents("bankno_error.log", $error, FILE_APPEND); 
  15. } 
  16. if( strlen($bankno) > 7 ){ 
  17. $sql = sprintf("update members set bankno = '%s' where id = '%s';", $bankno, $this->row['id']); 
  18. $this->sql = $sql; 
  19. } 
  20. printf("%s\n",$this->sql); 
  21. } 
  22. } 
  23. class Pool { 
  24. public $pool = array(); 
  25. public function __construct($count) { 
  26. $this->count = $count; 
  27. } 
  28. public function push($row){ 
  29. if(count($this->pool) < $this->count){ 
  30. $this->pool[] = new Update($row); 
  31. return true; 
  32. }else{ 
  33. return false; 
  34. } 
  35. } 
  36. public function start(){ 
  37. foreach ( $this->pool as $id => $worker){ 
  38. $this->pool[$id]->start(); 
  39. } 
  40. } 
  41. public function join(){ 
  42. foreach ( $this->pool as $id => $worker){ 
  43. $this->pool[$id]->join(); 
  44. } 
  45. } 
  46. public function clean(){ 
  47. foreach ( $this->pool as $id => $worker){ 
  48. if(! $worker->isRunning()){ 
  49. unset($this->pool[$id]); 
  50. } 
  51. } 
  52. } 
  53. } 
  54. try { 
  55. $dbh = new PDO("mysql:host=" . str_replace(':', ';port=', $dbhost) . ";dbname=$dbname", $dbuser, $dbpw, array( 
  56. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF8\'', 
  57. PDO::MYSQL_ATTR_COMPRESS => true 
  58. ) 
  59. ); 
  60. $sql = "select id,bankno from members order by id desc"; 
  61. $row = $dbh->query($sql); 
  62. $pool = new Pool(5); 
  63. while($member = $row->fetch(PDO::FETCH_ASSOC)) 
  64. { 
  65. while(true){ 
  66. if($pool->push($member)){ //压入任务到池中 
  67. break; 
  68. }else{ //如果池已经满，就开始启动线程 
  69. $pool->start(); 
  70. $pool->join(); 
  71. $pool->clean(); 
  72. } 
  73. } 
  74. } 
  75. $pool->start(); 
  76. $pool->join(); 
  77. $dbh = null; 
  78. } catch (Exception $e) { 
  79. echo '[' , date('H:i:s') , ']', '系统错误', $e->getMessage(), "\n"; 
  80. } 
  81. ?> 

动态队列线程池

上面的例子是当线程池满后执行start统一启动，下面的例子是只要线程池中有空闲便立即创建新线程。

  1. <?php 
  2. class Update extends Thread { 
  3. public $running = false; 
  4. public $row = array(); 
  5. public function __construct($row) { 
  6. $this->row = $row; 
  7. $this->sql = null; 
  8. //print_r($this->row); 
  9. } 
  10. public function run() { 
  11. if(strlen($this->row['bankno']) > 100 ){ 
  12. $bankno = safenet_decrypt($this->row['bankno']); 
  13. }else{ 
  14. $error = sprintf("%s, %s\r\n",$this->row['id'], $this->row['bankno']); 
  15. file_put_contents("bankno_error.log", $error, FILE_APPEND); 
  16. } 
  17. if( strlen($bankno) > 7 ){ 
  18. $sql = sprintf("update members set bankno = '%s' where id = '%s';", $bankno, $this->row['id']); 
  19. $this->sql = $sql; 
  20. } 
  21. printf("%s\n",$this->sql); 
  22. } 
  23. } 
  24. try { 
  25. $dbh = new PDO("mysql:host=" . str_replace(':', ';port=', $dbhost) . ";dbname=$dbname", $dbuser, $dbpw, array( 
  26. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF8\'', 
  27. PDO::MYSQL_ATTR_COMPRESS => true 
  28. ) 
  29. ); 
  30. $sql = "select id,bankno from members order by id desc limit 50"; 
  31. $row = $dbh->query($sql); 
  32. $pool = array(); 
  33. while($member = $row->fetch(PDO::FETCH_ASSOC)) 
  34. { 
  35. $id = $member['id']; 
  36. while (true){ 
  37. if(count($pool) < 5){ 
  38. $pool[$id] = new Update($member); 
  39. $pool[$id]->start(); 
  40. break; 
  41. }else{ 
  42. foreach ( $pool as $name => $worker){ 
  43. if(! $worker->isRunning()){ 
  44. unset($pool[$name]); 
  45. } 
  46. } 
  47. } 
  48. } 
  49. } 
  50. $dbh = null; 
  51. } catch (Exception $e) { 
  52. echo '【' , date('H:i:s') , '】', '【系统错误】', $e->getMessage(), "\n"; 
  53. } 
  54. ?> 

pthreads Pool类

  1. <?php 
  2. class WebWorker extends Worker { 
  3. public function __construct(SafeLog $logger) { 
  4. $this->logger = $logger; 
  5. } 
  6. protected $loger; 
  7. } 
  8. class WebWork extends Stackable { 
  9. public function isComplete() { 
  10. return $this->complete; 
  11. } 
  12. public function run() { 
  13. $this->worker 
  14. ->logger 
  15. ->log("%s executing in Thread #%lu", 
  16. __CLASS__, $this->worker->getThreadId()); 
  17. $this->complete = true; 
  18. } 
  19. protected $complete; 
  20. } 
  21. class SafeLog extends Stackable { 
  22. protected function log($message, $args = []) { 
  23. $args = func_get_args(); 
  24. if (($message = array_shift($args))) { 
  25. echo vsprintf( 
  26. "{$message}\n", $args); 
  27. } 
  28. } 
  29. } 
  30. $pool = new Pool(8, \WebWorker::class, [new SafeLog()]); 
  31. $pool->submit($w=new WebWork()); 
  32. $pool->submit(new WebWork()); 
  33. $pool->submit(new WebWork()); 
  34. $pool->submit(new WebWork()); 
  35. $pool->submit(new WebWork()); 
  36. $pool->submit(new WebWork()); 
  37. $pool->submit(new WebWork()); 
  38. $pool->submit(new WebWork()); 
  39. $pool->submit(new WebWork()); 
  40. $pool->submit(new WebWork()); 
  41. $pool->submit(new WebWork()); 
  42. $pool->submit(new WebWork()); 
  43. $pool->submit(new WebWork()); 
  44. $pool->submit(new WebWork()); 
  45. $pool->shutdown(); 
  46. $pool->collect(function($work){ 
  47. return $work->isComplete(); 
  48. }); 
  49. var_dump($pool); 

七、多线程文件安全读写

LOCK_SH 取得共享锁定（读取的程序）

LOCK_EX 取得独占锁定（写入的程序

LOCK_UN 释放锁定（无论共享或独占）

LOCK_NB 如果不希望 flock() 在锁定时堵塞

  1. <?php 
  2. $fp = fopen("/tmp/lock.txt", "r+"); 
  3. if (flock($fp, LOCK_EX)) { // 进行排它型锁定 
  4. ftruncate($fp, 0); // truncate file 
  5. fwrite($fp, "Write something here\n"); 
  6. fflush($fp); // flush output before releasing the lock 
  7. flock($fp, LOCK_UN); // 释放锁定 
  8. } else { 
  9. echo "Couldn't get the lock!"; 
  10. } 
  11. fclose($fp); 
  12. $fp = fopen('/tmp/lock.txt', 'r+'); 
  13. if(!flock($fp, LOCK_EX | LOCK_NB)) { 
  14. echo 'Unable to obtain lock'; 
  15. exit(-1); 
  16. } 
  17. fclose($fp); 
  18. ?> 

八、多线程与数据连接

pthreads 与 pdo 同时使用是，需要注意一点，需要静态声明public static $dbh;并且通过单例模式访问数据库连接。

Worker 与 PDO

  1. <?php 
  2. class Work extends Stackable { 
  3. public function __construct() { 
  4. } 
  5. public function run() { 
  6. $dbh = $this->worker->getConnection(); 
  7. $sql = "select id,name from members order by id desc limit "; 
  8. $row = $dbh->query($sql); 
  9. while($member = $row->fetch(PDO::FETCH_ASSOC)){ 
  10. print_r($member); 
  11. } 
  12. } 
  13. } 
  14. class ExampleWorker extends Worker { 
  15. public static $dbh; 
  16. public function __construct($name) { 
  17. } 
  18. /* 
  19. * The run method should just prepare the environment for the work that is coming ... 
  20. */ 
  21. public function run(){ 
  22. self::$dbh = new PDO('mysql:host=...;dbname=example','www',''); 
  23. } 
  24. public function getConnection(){ 
  25. return self::$dbh; 
  26. } 
  27. } 
  28. $worker = new ExampleWorker("My Worker Thread"); 
  29. $work=new Work(); 
  30. $worker->stack($work); 
  31. $worker->start(); 
  32. $worker->shutdown(); 
  33. ?> 

Pool 与 PDO

在线程池中链接数据库

  1. # cat pool.php 
  2. <?php 
  3. class ExampleWorker extends Worker { 
  4. public function __construct(Logging $logger) { 
  5. $this->logger = $logger; 
  6. } 
  7. protected $logger; 
  8. } 
  9. /* the collectable class implements machinery for Pool::collect */ 
  10. class Work extends Stackable { 
  11. public function __construct($number) { 
  12. $this->number = $number; 
  13. } 
  14. public function run() { 
  15. $dbhost = 'db.example.com'; // 数据库服务器 
  16. $dbuser = 'example.com'; // 数据库用户名 
  17. $dbpw = 'password'; // 数据库密码 
  18. $dbname = 'example_real'; 
  19. $dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array( 
  20. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF\'', 
  21. PDO::MYSQL_ATTR_COMPRESS => true, 
  22. PDO::ATTR_PERSISTENT => true 
  23. ) 
  24. ); 
  25. $sql = "select OPEN_TIME, `COMMENT` from MT_TRADES where LOGIN='".$this->number['name']."' and CMD='' and `COMMENT` = '".$this->number['order'].":DEPOSIT'"; 
  26. #echo $sql; 
  27. $row = $dbh->query($sql); 
  28. $mt_trades = $row->fetch(PDO::FETCH_ASSOC); 
  29. if($mt_trades){ 
  30. $row = null; 
  31. $sql = "UPDATE db_example.accounts SET paystatus='成功', deposit_time='".$mt_trades['OPEN_TIME']."' where `order` = '".$this->number['order']."';"; 
  32. $dbh->query($sql); 
  33. #printf("%s\n",$sql); 
  34. } 
  35. $dbh = null; 
  36. printf("runtime: %s, %s, %s\n", date('Y-m-d H:i:s'), $this->worker->getThreadId() ,$this->number['order']); 
  37. } 
  38. } 
  39. class Logging extends Stackable { 
  40. protected static $dbh; 
  41. public function __construct() { 
  42. $dbhost = 'db.example.com'; // 数据库服务器 
  43. $dbuser = 'example.com'; // 数据库用户名 
  44. $dbpw = 'password'; // 数据库密码 
  45. $dbname = 'example_real'; // 数据库名 
  46. self::$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array( 
  47. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF\'', 
  48. PDO::MYSQL_ATTR_COMPRESS => true 
  49. ) 
  50. ); 
  51. } 
  52. protected function log($message, $args = []) { 
  53. $args = func_get_args(); 
  54. if (($message = array_shift($args))) { 
  55. echo vsprintf("{$message}\n", $args); 
  56. } 
  57. } 
  58. protected function getConnection(){ 
  59. return self::$dbh; 
  60. } 
  61. } 
  62. $pool = new Pool(, \ExampleWorker::class, [new Logging()]); 
  63. $dbhost = 'db.example.com'; // 数据库服务器 
  64. $dbuser = 'example.com'; // 数据库用户名 
  65. $dbpw = 'password'; // 数据库密码 
  66. $dbname = 'db_example'; 
  67. $dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array( 
  68. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF\'', 
  69. PDO::MYSQL_ATTR_COMPRESS => true 
  70. ) 
  71. ); 
  72. $sql = "select `order`,name from accounts where deposit_time is null order by id desc"; 
  73. $row = $dbh->query($sql); 
  74. while($account = $row->fetch(PDO::FETCH_ASSOC)) 
  75. { 
  76. $pool->submit(new Work($account)); 
  77. } 
  78. $pool->shutdown(); 
  79. ?> 

进一步改进上面程序，我们使用单例模式 $this->worker->getInstance(); 全局仅仅做一次数据库连接，线程使用共享的数据库连接

  1. <?php 
  2. class ExampleWorker extends Worker { 
  3. #public function __construct(Logging $logger) { 
  4. # $this->logger = $logger; 
  5. #} 
  6. #protected $logger; 
  7. protected static $dbh; 
  8. public function __construct() { 
  9. } 
  10. public function run(){ 
  11. $dbhost = 'db.example.com'; // 数据库服务器 
  12. $dbuser = 'example.com'; // 数据库用户名 
  13. $dbpw = 'password'; // 数据库密码 
  14. $dbname = 'example'; // 数据库名 
  15. self::$dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array( 
  16. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF\'', 
  17. PDO::MYSQL_ATTR_COMPRESS => true, 
  18. PDO::ATTR_PERSISTENT => true 
  19. ) 
  20. ); 
  21. } 
  22. protected function getInstance(){ 
  23. return self::$dbh; 
  24. } 
  25. } 
  26. /* the collectable class implements machinery for Pool::collect */ 
  27. class Work extends Stackable { 
  28. public function __construct($data) { 
  29. $this->data = $data; 
  30. #print_r($data); 
  31. } 
  32. public function run() { 
  33. #$this->worker->logger->log("%s executing in Thread #%lu", __CLASS__, $this->worker->getThreadId() ); 
  34. try { 
  35. $dbh = $this->worker->getInstance(); 
  36. #print_r($dbh); 
  37. $id = $this->data['id']; 
  38. $mobile = safenet_decrypt($this->data['mobile']); 
  39. #printf("%d, %s \n", $id, $mobile); 
  40. if(strlen($mobile) > ){ 
  41. $mobile = substr($mobile, -); 
  42. } 
  43. if($mobile == 'null'){ 
  44. # $sql = "UPDATE members_digest SET mobile = '".$mobile."' where id = '".$id."'"; 
  45. # printf("%s\n",$sql); 
  46. # $dbh->query($sql); 
  47. $mobile = ''; 
  48. $sql = "UPDATE members_digest SET mobile = :mobile where id = :id"; 
  49. }else{ 
  50. $sql = "UPDATE members_digest SET mobile = md(:mobile) where id = :id"; 
  51. } 
  52. $sth = $dbh->prepare($sql); 
  53. $sth->bindValue(':mobile', $mobile); 
  54. $sth->bindValue(':id', $id); 
  55. $sth->execute(); 
  56. #echo $sth->debugDumpParams(); 
  57. } 
  58. catch(PDOException $e) { 
  59. $error = sprintf("%s,%s\n", $mobile, $id ); 
  60. file_put_contents("mobile_error.log", $error, FILE_APPEND); 
  61. } 
  62. #$dbh = null; 
  63. printf("runtime: %s, %s, %s, %s\n", date('Y-m-d H:i:s'), $this->worker->getThreadId() ,$mobile, $id); 
  64. #printf("runtime: %s, %s\n", date('Y-m-d H:i:s'), $this->number); 
  65. } 
  66. } 
  67. $pool = new Pool(, \ExampleWorker::class, []); 
  68. #foreach (range(, ) as $number) { 
  69. # $pool->submit(new Work($number)); 
  70. #} 
  71. $dbhost = 'db.example.com'; // 数据库服务器 
  72. $dbuser = 'example.com'; // 数据库用户名 
  73. $dbpw = 'password'; // 数据库密码 
  74. $dbname = 'example'; 
  75. $dbh = new PDO("mysql:host=$dbhost;port=;dbname=$dbname", $dbuser, $dbpw, array( 
  76. PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES \'UTF\'', 
  77. PDO::MYSQL_ATTR_COMPRESS => true 
  78. ) 
  79. ); 
  80. #print_r($dbh); 
  81. #$sql = "select id, mobile from members where id < :id"; 
  82. #$sth = $dbh->prepare($sql); 
  83. #$sth->bindValue(':id',); 
  84. #$sth->execute(); 
  85. #$result = $sth->fetchAll(); 
  86. #print_r($result); 
  87. # 
  88. #$sql = "UPDATE members_digest SET mobile = :mobile where id = :id"; 
  89. #$sth = $dbh->prepare($sql); 
  90. #$sth->bindValue(':mobile', 'aa'); 
  91. #$sth->bindValue(':id',''); 
  92. #echo $sth->execute(); 
  93. #echo $sth->queryString; 
  94. #echo $sth->debugDumpParams(); 
  95. $sql = "select id, mobile from members order by id asc"; // limit "; 
  96. $row = $dbh->query($sql); 
  97. while($members = $row->fetch(PDO::FETCH_ASSOC)) 
  98. { 
  99. #$order = $account['order']; 
  100. #printf("%s\n",$order); 
  101. //print_r($members); 
  102. $pool->submit(new Work($members)); 
  103. #unset($account['order']); 
  104. } 
  105. $pool->shutdown(); 
  106. ?> 

多线程中操作数据库总结

总的来说 pthreads 仍然处在发展中，仍有一些不足的地方，我们也可以看到pthreads的git在不断改进这个项目

数据库持久链接很重要，否则每个线程都会开启一次数据库连接，然后关闭，会导致很多链接超时。

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

Tags: pthreads php多线程安装

分享到：

相关文章

推荐文章

热门文章

最新评论文章