php stream_select兑现可控并发

<?php

$time = 100;
$parallel = 10;
$sockets = array();
$results = array();
$failed = array();

$id = 0;
$url = 'localhost:8080';
$http_msg = "GET /stream_select_req.php?delay=%s";
$http_msg .= "HTTP/1.0\r\n";
$http_msg .= "Host:localhost:8080\r\n\r\n";

$timeout = 300;
for($i = 0 ; $i < $parallel ; $i++){
        $s = stream_socket_client($url,$errno,$errstr,$timeout,STREAM_CLIENT_ASYNC_CONNECT|STREAM_CLIENT_CONNECT);
        if($s){
                $sockets[$id++] = $s;
                fwrite($s,sprintf($http_msg,$i));
        }else{
                echo "Stream $id falied to open correctly.";
                $failed[] = sprintf($http_msg,$i);
        }
}

while(count($sockets)){
        $read = $sockets;
        stream_select($read,$w=null,$e=null,$timeout);
        if(count($read)){
                foreach($read as $r){
                        $index = array_search($r,$sockets);
                        $data = fread($r,1024);
                        if(strlen($data) == 0 ){
                                echo "Stream $index closes at ".date('h:i:s')."<br>";
                                fclose($r);
                                unset($sockets[$index]);
                                //add new task
                                $next = $id++;
                                if($next < $time){
                                        $s = stream_socket_client($url,$errno,$errstr,$timeout,STREAM_CLIENT_ASYNC_CONNECT|STREAM_CLIENT_CONNECT);
                                        $sockets[$next] = $s;
                                        fwrite($s,sprintf($http_msg,$id));
                                }
                        }
                }
        }
}





?>