A summary of several ways to try program concurrency in PHP programming, php programming

This article roughly summarizes the five concurrency methods in PHP programming:
1.curl_multi_init
The document says Allows the processing of multiple cURL handles asynchronously. It is indeed asynchronous. What needs to be understood here is the select method, which is explained in the document Blocks until there is activity on any of the curl_multi connections.. You should be able to understand if you understand the common asynchronous models, select and epoll are all famous

<&#63;php
// build the individual requests as above, but do not execute them
$ch_1 = curl_init('http://www.bkjia.com/');
$ch_2 = curl_init('http://www.bkjia.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);

What I set here is that when the select gets the result, it will exit the loop and delete the curl resource, so as to cancel the http request.

2.swoole_client
swoole_client provides asynchronous mode, I actually forgot about this. The sleep method here requires that the swoole version is greater than or equal to 1.7.21. I haven't upgraded to this version yet, so I can just exit directly.

<&#63;php
$client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
//設(shè)置事件回調(diào)函數(shù)
$client->on("connect", function($cli) {
  $req = "GET / HTTP/1.1\r\n
  Host: www.jb51.net\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";
});
//發(fā)起網(wǎng)絡(luò)連接
$client->connect('183.207.95.145', 80, 1);

3.process
Hey, I almost forgot about swoole_process. There is no need for the pcntl module here. But after writing it, I found that this is not actually an interrupt request, but whichever comes first is read, ignoring the subsequent return value.

<&#63;php

$workers = [];
$worker_num = 3;//創(chuàng)建的進(jìn)程數(shù)
$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){
  //子進(jìn)程也會(huì)包含此事件
  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
When compiling the pthreads module, it prompts that ZTS must be turned on when compiling PHP, so it seems that the thread safe version must be used. Many PHP in wamp happens to be TS, so I directly downloaded a dll and copied the instructions in the document to the corresponding directory. Tested under win. I still don’t fully understand it, but I found an article saying that PHP’s pthreads and POSIX pthreads are completely different. The code is a bit bad, I need to read more documentation to get a feel for it.

<&#63;php
class Foo extends Stackable {
  public $url;
  public $response = null;
  public function __construct(){
    $this->url = 'http://www.bkjia.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
Write asynchronous code in a synchronous manner:

<&#63;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(); 
 
&#63;> 

Code interpretation:

Use PHP’s built-in array capability to implement simple “timeout management” and use millisecond precision as time slicing;
Encapsulate the AsyncSendRecv interface and call it in the form of yield AsyncSendRecv(), which is more natural;
Add Exception as an error handling mechanism. You can also add ret_code, which is for demonstration purposes only.

Articles you may be interested in:

• PHP session lock and concurrency
• php solves the problem of negative inventory caused by large-traffic concurrent warehousing such as flash sales and lottery draws
• php automatically generates thumbnails based on URLs and handles high concurrency issues
• Solution to concurrent read and write file conflicts in php
• Solution to the pressure that php concurrency puts on MYSQL
• A PHP concurrent access example code
• PHP curl concurrency best practice code sharing
• How PHP solves the problem of large website traffic and high concurrency
• Nginx PHP (FastCGI ) Build a high-concurrency WEB server (automatic installation script) second edition
• Common locking and locking PHP specific implementation codes under concurrency