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));                                }                        }                }        }}?>