社群 學習 工具庫 休閒
搜尋
繁体中文
首頁 > 後端開發 > php教程 > php pthreads多執行緒的安裝與使用

php pthreads多執行緒的安裝與使用

高洛峰
發布： 2023-03-03 19:42:01
原創
1394 人瀏覽過

安裝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
登入後複製

   

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 aredcuted Stackby Workers reads sreads？ 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(&#39;mysql:host=10.0.0.30;dbname=testdb&#39;,&#39;root&#39;,&#39;123456&#39;);
}
public function getConnection(){
return self::$dbh;
}
}
$worker = new ExampleWorker("My Worker Thread");
$sql1 = new SQLQuery(&#39;select * from test order by id desc limit 1,5&#39;);
$worker->stack($sql1);
$sql2 = new SQLQuery(&#39;select * from test order by id desc limit 5,5&#39;);
$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__, &#39;PHP&#39;);
$key = ftok($tmp, &#39;a&#39;);
$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__, &#39;PHP&#39;);
$key = ftok($tmp, &#39;a&#39;);
$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類

<?php
class 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[&#39;bankno&#39;]) > 100 ){
$bankno = safenet_decrypt($this->row[&#39;bankno&#39;]);
}else{
$error = sprintf("%s, %s\r\n",$this->row[&#39;id&#39;], $this->row[&#39;bankno&#39;]);
file_put_contents("bankno_error.log", $error, FILE_APPEND);
}
if( strlen($bankno) > 7 ){
$sql = sprintf("update members set bankno = &#39;%s&#39; where id = &#39;%s&#39;;", $bankno, $this->row[&#39;id&#39;]);
$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(&#39;:&#39;, &#39;;port=&#39;, $dbhost) . ";dbname=$dbname", $dbuser, $dbpw, array(
PDO::MYSQL_ATTR_INIT_COMMAND => &#39;SET NAMES \&#39;UTF8\&#39;&#39;,
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 &#39;[&#39; , date(&#39;H:i:s&#39;) , &#39;]&#39;, &#39;系统错误&#39;, $e->getMessage(), "\n";
}
?>
登入後複製

   


動態隊列線程池

上面的例子是當線程池滿後執行。 

<?php
class 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[&#39;bankno&#39;]) > 100 ){
$bankno = safenet_decrypt($this->row[&#39;bankno&#39;]);
}else{
$error = sprintf("%s, %s\r\n",$this->row[&#39;id&#39;], $this->row[&#39;bankno&#39;]);
file_put_contents("bankno_error.log", $error, FILE_APPEND);
}
if( strlen($bankno) > 7 ){
$sql = sprintf("update members set bankno = &#39;%s&#39; where id = &#39;%s&#39;;", $bankno, $this->row[&#39;id&#39;]);
$this->sql = $sql;
}
printf("%s\n",$this->sql);
}
}
try {
$dbh = new PDO("mysql:host=" . str_replace(&#39;:&#39;, &#39;;port=&#39;, $dbhost) . ";dbname=$dbname", $dbuser, $dbpw, array(
PDO::MYSQL_ATTR_INIT_COMMAND => &#39;SET NAMES \&#39;UTF8\&#39;&#39;,
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[&#39;id&#39;];
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 &#39;【&#39; , date(&#39;H:i:s&#39;) , &#39;】&#39;, &#39;【系统错误】&#39;, $e->getMessage(), "\n";
}
?>
登入後複製

   


pthreads Pool類

<?php
class 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);
登入後複製

   

七、多取文件安全讀寫程式取得的佔用空間編號入的程序

LOCK_UN 釋放鎖定（無論共享或獨佔）

LOCK_NB 如果不希望flock() 在鎖定時堵塞

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

   


、多肉與資料連接時需要注意使用資料連接器一點，需要靜態聲明public static $dbh;並且透過單例模式存取資料庫連接。

進一步改進上面程序，我們使用單例模式$this->worker->getInstance(); 全局僅做一次數據庫連接，線程使用共享的數據庫連接

<?php
class 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(&#39;mysql:host=...;dbname=example&#39;,&#39;www&#39;,&#39;&#39;);
}
public function getConnection(){
return self::$dbh;
}
}
$worker = new ExampleWorker("My Worker Thread");
$work=new Work();
$worker->stack($work);
$worker->start();
$worker->shutdown();
?>
登入後複製
   

多線程中操作資料庫總結

總的來說pthreads 仍然處在發展中，仍有一些不足的地方，我們也可以看到pthreads的git在不斷改進這個項目


數據庫持久鏈接很重要，否則每個線程都會開啟一次數據庫連接，然後關閉，會導致很多連結逾時。多執行緒的安裝與使用相關文章請關注PHP中文網！
相關標籤：
pthreads多线程
來源：php.cn
上一篇：php中foreach結合curl實作多執行緒的方法分析 下一篇：PHP使用Pthread實作的多執行緒操作實例
本網站聲明
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
作者最新文章
最新問題
Laravel 9.x中的常規錯誤：字段'id'在沒有預設值的情況下 我在我的應用程式中使用UUID，並且我已經實現了該特徵，如線上所示，如下所示：traitUuid{protectedstaticfunctionboot():void{parent...
來自於 2024-04-06 11:12:54
0
1
350
npx create-react-app 指令出現錯誤，我該如何解決？ 我一直在嘗試使用'npxcreate-react-app'命令創建一個新的React應用，但似乎對我來說無效。最近有其他人遇到這個問題嗎？我有一個穩定的互聯網連接，並確保我的系統上...
來自於 2024-04-05 14:42:18
0
1
3511
"waitForSelector"在元素可見前超時等待，儘管元素在螢幕上 我正在建造一個結帳機器人，但是在處理那些愚蠢的“輸入手機號碼以接收短信”的彈窗時遇到了麻煩，我告訴機器人要點擊的元素出現在屏幕上，但機器人卻不點擊它。這是我第一個使用JavaScr...
來自於 2024-04-04 18:31:08
0
1
1416
一個簡單的Hello world程式建立多個節點實例 我對Javascript和NodeJS非常陌生。我正在運行一個簡單的helloworld程序，如下方案1constdurationInSeconds=10;console.log(...
來自於 2024-04-04 11:03:22
0
1
310
視窗中的內容溢出阻止滾動並隱藏新內容 我目前正在建立一個輪盤系統程式（這更多的是為了阻止我下注而不是下注！）並且我只是在做基本框架，但已經遇到了主視窗“#results”不滾動的問題充滿了結果。捲動需要跟隨最新的內容行...
來自於 2024-04-03 21:24:55
0
1
285
相關專題
更多>
熱門推薦
熱門教學
更多>
相關教學
熱門推薦
最新課程
最新下載
更多>
網站特效
網站源碼
網站素材
前端模板