服务器之家：专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|JAVA教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|JavaScript|易语言|

服务器之家 - 编程语言 - PHP教程 - PHP编程中尝试程序并发的几种方式总结

PHP编程中尝试程序并发的几种方式总结

2021-01-03 17:13一堆好人卡 PHP教程

这篇文章主要介绍了PHP编程中尝试程序并发的几种方式总结,这里举了借助yield的异步以及swoole_process的进程创建等例子,PHP本身并不支持多线程并发,需要的朋友可以参考下

本文大约总结了PHP编程中的五种并发方式：
1.curl_multi_init
文档中说的是 Allows the processing of multiple cURL handles asynchronously. 确实是异步。这里需要理解的是select这个方法，文档中是这么解释的Blocks until there is activity on any of the curl_multi connections.。了解一下常见的异步模型就应该能理解，select, epoll，都很有名

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
<?php
// build the individual requests as above, but do not execute them
$ch_1 = curl_init('http://www.zzvips.com/');
$ch_2 = curl_init('http://www.zzvips.com/');
curl_setopt($ch_1, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch_2, CURLOPT_RETURNTRANSFER, true);
 
// build the multi-curl handle, adding both $ch
$mh = curl_multi_init();
curl_multi_add_handle($mh, $ch_1);
curl_multi_add_handle($mh, $ch_2);
 
// execute all queries simultaneously, and continue when all are complete
$running = null;
do {
  curl_multi_exec($mh, $running);
  $ch = curl_multi_select($mh);
  if($ch !== 0){
    $info = curl_multi_info_read($mh);
    if($info){
      var_dump($info);
      $response_1 = curl_multi_getcontent($info['handle']);
      echo "$response_1 \n";
      break;
    }
  }
} while ($running > 0);
 
//close the handles
curl_multi_remove_handle($mh, $ch_1);
curl_multi_remove_handle($mh, $ch_2);
curl_multi_close($mh);

这里我设置的是，select得到结果，就退出循环，并且删除 curl resource, 从而达到取消http请求的目的。

2.swoole_client
swoole_client提供了异步模式，我竟然把这个忘了。这里的sleep方法需要swoole版本大于等于1.7.21, 我还没升到这个版本，所以直接exit也可以。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
<?php
$client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
//设置事件回调函数
$client->on("connect", function($cli) {
  $req = "GET / HTTP/1.1\r\n
  Host: www.zzvips.com\r\n
  Connection: keep-alive\r\n
  Cache-Control: no-cache\r\n
  Pragma: no-cache\r\n\r\n";
 
  for ($i=0; $i < 3; $i++) {
    $cli->send($req);
  }
});
$client->on("receive", function($cli, $data){
  echo "Received: ".$data."\n";
  exit(0);
  $cli->sleep(); // swoole >= 1.7.21
});
$client->on("error", function($cli){
  echo "Connect failed\n";
});
$client->on("close", function($cli){
  echo "Connection close\n";
});
//发起网络连接
$client->connect('183.207.95.145', 80, 1);

3.process
哎，竟然差点忘了 swoole_process, 这里就不用 pcntl 模块了。但是写完发现，这其实也不算是中断请求，而是哪个先到读哪个，忽视后面的返回值。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
<?php
 
$workers = [];
$worker_num = 3;//创建的进程数
$finished = false;
$lock = new swoole_lock(SWOOLE_MUTEX);
 
for($i=0;$i<$worker_num ; $i++){
  $process = new swoole_process('process');
  //$process->useQueue();
  $pid = $process->start();
  $workers[$pid] = $process;
}
 
foreach($workers as $pid => $process){
  //子进程也会包含此事件
  swoole_event_add($process->pipe, function ($pipe) use($process, $lock, &$finished) {
    $lock->lock();
    if(!$finished){
      $finished = true;
      $data = $process->read();
      echo "RECV: " . $data.PHP_EOL;
    }
    $lock->unlock();
  });
}
 
function process(swoole_process $process){
  $response = 'http response';
  $process->write($response);
  echo $process->pid,"\t",$process->callback .PHP_EOL;
}
 
for($i = 0; $i < $worker_num; $i++) {
  $ret = swoole_process::wait();
  $pid = $ret['pid'];
  echo "Worker Exit, PID=".$pid.PHP_EOL;
}

4.pthreads
编译pthreads模块时，提示php编译时必须打开ZTS, 所以貌似必须 thread safe 版本才能使用. wamp中多php正好是TS的，直接下了个dll, 文档中的说明复制到对应目录，就在win下测试了。 还没完全理解，查到文章说 php 的 pthreads 和 POSIX pthreads是完全不一样的。代码有些烂，还需要多看看文档，体会一下。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
<?php
class Foo extends Stackable {
  public $url;
  public $response = null;
  public function __construct(){
    $this->url = 'http://www.zzvips.com';
  }
  public function run(){}
}
 
class Process extends Worker {
  private $text = "";
  public function __construct($text,$object){
    $this->text = $text;
    $this->object = $object;
  }
  public function run(){
    while (is_null($this->object->response)){
      print " Thread {$this->text} is running\n";
      $this->object->response = 'http response';
      sleep(1);
    }
  }
}
 
$foo = new Foo();
 
$a = new Process("A",$foo);
$a->start();
 
$b = new Process("B",$foo);
$b->start();
 
echo $foo->response;

5.yield
以同步方式书写异步代码：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
<?php
 
class AsyncServer {
  protected $handler;
  protected $socket;
  protected $tasks = [];
  protected $timers = [];
 
  public function __construct(callable $handler) {
    $this->handler = $handler;
 
    $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
    if(!$this->socket) {
      die(socket_strerror(socket_last_error())."\n");
    }
    if (!socket_set_nonblock($this->socket)) {
      die(socket_strerror(socket_last_error())."\n");
    }
    if(!socket_bind($this->socket, "0.0.0.0", 1234)) {
      die(socket_strerror(socket_last_error())."\n");
    }
  }
 
  public function Run() {
    while (true) {
      $now = microtime(true) * 1000;
      foreach ($this->timers as $time => $sockets) {
        if ($time > $now) break;
        foreach ($sockets as $one) {
          list($socket, $coroutine) = $this->tasks[$one];
          unset($this->tasks[$one]);
          socket_close($socket);
          $coroutine->throw(new Exception("Timeout"));
        }
        unset($this->timers[$time]);
      }
 
      $reads = array($this->socket);
      foreach ($this->tasks as list($socket)) {
        $reads[] = $socket;
      }
      $writes = NULL;
      $excepts= NULL;
      if (!socket_select($reads, $writes, $excepts, 0, 1000)) {
        continue;
      }
 
      foreach ($reads as $one) {
        $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);
        if (!$len) {
          //echo "socket_recvfrom fail.\n";
          continue;
        }
        if ($one == $this->socket) {
          //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n";
          $handler = $this->handler;
          $coroutine = $handler($one, $data, $len, $ip, $port);
          if (!$coroutine) {
            //echo "[Run]everything is done.\n";
            continue;
          }
          $task = $coroutine->current();
          //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout\n";
          $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
          if(!$socket) {
            //echo socket_strerror(socket_last_error())."\n";
            $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
            continue;
          }
          if (!socket_set_nonblock($socket)) {
            //echo socket_strerror(socket_last_error())."\n";
            $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
            continue;
          }
          socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);
          $deadline = $now + $task->timeout;
          $this->tasks[$socket] = [$socket, $coroutine, $deadline];
          $this->timers[$deadline][$socket] = $socket;
        } else {
          //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port\n";
          list($socket, $coroutine, $deadline) = $this->tasks[$one];
          unset($this->tasks[$one]);
          unset($this->timers[$deadline][$one]);
          socket_close($socket);
          $coroutine->send(array($data, $len));
        }
      }
    }
  }
}
 
class AsyncTask {
  public $data;
  public $len;
  public $ip;
  public $port;
  public $timeout;
 
  public function __construct($data, $len, $ip, $port, $timeout) {
    $this->data = $data;
    $this->len = $len;
    $this->ip = $ip;
    $this->port = $port;
    $this->timeout = $timeout;
  }
}
 
function AsyncSendRecv($req_buf, $req_len, $ip, $port, $timeout) {
  return new AsyncTask($req_buf, $req_len, $ip, $port, $timeout);
}
 
function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {
  //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf\n";
  try {
    list($rsp_buf, $rsp_len) = (yield AsyncSendRecv($req_buf, $req_len, "127.0.0.1", 2345, 3000));
  } catch (Exception $ex) {
    $rsp_buf = $ex->getMessage();
    $rsp_len = strlen($rsp_buf);
    //echo "[Exception]$rsp_buf\n";
  }
  //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf\n";
  socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);
}
 
$server = new AsyncServer(RequestHandler);
$server->Run();
 
?>

代码解读：

借助PHP内置array能力，实现简单的“超时管理”，以毫秒为精度作为时间分片；
封装AsyncSendRecv接口，调用形如yield AsyncSendRecv()，更加自然；
添加Exception作为错误处理机制，添加ret_code亦可，仅为展示之用。

  • PHP
  • 并发

    • 延伸 · 阅读

    精彩推荐
    最近更新
    编辑推荐
    阅读排行
    热门标签
    2021 　 2038 　 100010 　 在线时间 　 bcompiler 　 PHP文件 　 形参 　 实参 　 存储方案 　 PHP技术 　 php安全 　 编程效率 　 Foxmail 　 PHPnow 　 apache_pn 　 visitFile 　 dede模板标签 　 ExcelFileParser 　 escape 　 IP查询 　 GOOGLE地图 　 递归循环 　 异步队列 　 颠倒 　 变量引用 　 自动获取 　 递归删除 　 加载文件 　 异常机制 　 递增运算 　
    © 2019-2020 服务器之家 版权所有 www.zzvips.com 关于我们联系我们版权申明网站地图