socket_select · PHP筆記

## **socket_select**(array **&$read**,array **&$write**,array **&$except**,int **$tv_sec** [,int **$tv_usec** = 0] ) :int 作用：獲取read數組中活動的socket，并且把不活躍的從read數組中刪除接受套接字組成的數組 **$read**，并等待它們更改狀態（就是**有新消息**到或者有客戶端**連接**/**斷開**時）在套接字數組 $read 中最初應保有一個服務端監聽套接字，每當該套接字可讀時，就表示有一個用戶發起了連接。此時你需要對該連接創建一個套接字，并加入到 $read 數組中除了這個服務端監聽套接字，客戶端監聽的套接字會變成可讀的，用戶套接字也會變成可讀的，此時你就可以讀取用戶發來的數據了 select處于等待時,兩個客戶端中甲先發數據來,則socket_select會在readfds中保留甲的socket并往下運行,另一個客戶端的socket就被丟棄了,所以再次循環時,變成只監聽甲了,這個可以在新循環中把所有鏈接的客戶端socket再次加進readfds中,則可以避免本程序的這個邏輯錯誤 **返回值**：＜0表示有錯誤，其他值表示可操作的socket個數第一個參數傳入要檢查的socket數組，函數返回后會變成包含可讀取消息的socket的數組，第二百個和第三個是可寫入(發送)數據的，第三個是出錯的第四個參數如果是null，表示知道出現可操作的socket，否則會持續阻塞度。0表示立即返回，其他值表示最多等待指定的秒數后返回這是一個同步方法，必須得到響應之后才會繼續下一步,常用在同步非阻塞IO 說明: * 1?新連接到來時,被監聽的端口是活躍的,如果是新數據到來或者客戶端關閉鏈接時,活躍的是對應的客戶端socket而不是服務器上被監聽的端口 * 2?如果客戶端發來數據沒有被讀走,則socket_select將會始終顯示客戶端是活躍狀態并將其保存在readfds數組中 * 3?如果客戶端先關閉了,則必須手動關閉服務器上相對應的客戶端socket,否則socket_select也始終顯示該客戶端活躍(這個道理跟"有新連接到來然后沒有用socket_access把它讀出來,導致監聽的端口一直活躍"是一樣的) 這個函數是同時接受多個連接的關鍵，我的理解它是為了阻塞程序繼續往下執行和自動選擇當前有活動的連接。 ``` $read = array($socket1, $socket2); $write = NULL; $except = NULL; $num_changed_sockets = socket_select($read, $write, $except, 0); if ($num_changed_sockets === false) { /* 錯誤處理 */ socket_last_error(); } else if ($num_changed_sockets > 0) { /*至少有一個套接字發生改變 */ } ``` ## **accept與select區別：** 一般經過創建套接字socket_socket（）綁定socket_bind（）以及socket_listen（）之后， ?就可以調用socket_select查看指定socket的狀態和 socket_accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen)建立一個全新連接; 判斷是否有客戶端發起鏈接請求， ? 一般用select（），然后accept（）。應該先select，可以測一下超時。如果先accept，select就沒必要了。還有就是select后，可以測一下資源，如果fd用完了，可以等歸還。直接的accept，發生這事這個連接就丟了。 ———————————————— ``` $readfds = array(); $writefds = array(); $sock = socket_create_listen(2000); //socket_set_nonblock($sock); // 非阻塞 //echo "sleep 10 second...\n"; //sleep(10); socket_getsockname($sock, $addr, $port); print "Server Listening on $addr:$port\n"; $readfds[(int)$sock]=$sock; $conn=socket_accept($sock); $readfds[]=$conn; $conn=socket_accept($sock); $readfds[]=$conn; $e = null; $t=100; $i=1; while(true){ echo "No.$i\n"; //當select處于等待時,兩個客戶端中甲先發數據來,則socket_select會在readfds中保留甲的socket并往下運行,另一個客戶端的socket就被丟棄了,所以再次循環時,變成只監聽甲了,這個可以在新循環中把所有鏈接的客戶端socket再次加進readfds中,則可以避免本程序的這個邏輯錯誤 echo @socket_select($readfds, $writefds, $e, $t)."\n"; var_dump($readfds); if(in_array($sock, $readfds)){ echo "2000 port is activity"; $readfds[]=socket_accept($sock); } //將讀取到的資源輸出 foreach ($readfds as $s){ if($s!=$sock){ //新連接到來時,被監聽的端口是活躍的,如果是新數據到來或者客戶端關閉鏈接時,活躍的是對應的客戶端socket而不是服務器上被監聽的端口 //如果客戶端發來數據沒有被讀走,則socket_select將會始終顯示客戶端是活躍狀態并將其保存在readfds數組中 //如果客戶端先關閉了,則必須手動關閉服務器上相對應的客戶端socket,否則socket_select也始終顯示該客戶端活躍(這個道理跟"有新連接到來然后沒有用socket_access把它讀出來,導致監聽的端口一直活躍"是一樣的) $result=@socket_read($s, 1024,PHP_NORMAL_READ); if($result===false){ $err_code=socket_last_error(); $err_test=socket_strerror($err_code); echo "client ".(int)$s." has closed[$err_code:$err_test]\n"; //手動關閉客戶端,最好清除一下$readfds數組中對應的元素 socket_shutdown($s); socket_close($s); }else{ echo $result; } } } usleep(3000000); $readfds[(int)$sock]=$sock; $i++; } //while無限循環，執行不到下面的代碼 ``` 示例 ``` $port = 9050; // 創建類型為TCP / IP的流套接字 $sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP); // 設置選項以重用端口 socket_set_option($sock, SOL_SOCKET, SO_REUSEADDR, 1); // 將套接字綁定到端口localhost地址的$port上 // 這意味著此端口上的所有連接現在都由我們負責發送/接收數據、斷開連接等 socket_bind($sock, 0, $port); // 開始監聽鏈接 socket_listen($sock); // 創建將與我們連接的所有客戶端的列表 // 將偵聽套接字添加到此列表 $clients = array($sock); while (true) { // 創建一個副本，這樣$clients不會被socket_select修改 $read = $clients; // 獲取所有要讀取數據的客戶端的列表 // 如果客戶端沒有數據，跳過此次循環轉到下一個迭代 if (socket_select($read, $write = NULL, $except = NULL, 0) < 1) continue; // 檢查是否有客戶端嘗試連接 if (in_array($sock, $read)) { // 接受客戶端的鏈接，并將其套接字添加到$clients數組中 $newsock = socket_accept($sock); $clients[] = $newsock; // 向客戶發送歡迎信息 socket_write($newsock, "目前有".(count($clients) - 1)." 個客戶端鏈接到服務器\n"); //查詢給定套接字的遠程端，可能返回主機端口或者Unix文件系統路徑（取決于其類型） socket_getpeername($newsock, $ip); echo "新的客戶端連接: {$ip}\n"; // 從帶數據的客戶端數組中刪除偵聽套接字 $key = array_search($sock, $read); unset($read[$key]); } // 遍歷所有要讀取數據的客戶端 foreach ($read as $read_sock) { // 一直讀到換行符或1024字節 // 客戶端斷開連接時，socket_read會顯示錯誤，因此請忽略錯誤消息 $data = @socket_read($read_sock, 1024, PHP_NORMAL_READ); // 檢查客戶端是否斷開連接 if ($data === false) { // 刪除$clients數組的客戶端 $key = array_search($read_sock, $clients); unset($clients[$key]); echo "client disconnected.\n"; // 繼續下一個要讀取的客戶端（如果有） continue; } // 刪除尾部/開頭的空白 $data = trim($data); // 檢查刪除空格后是否有任何數據 if (!empty($data)) { // 將此消息發送到$CLIENTS數組中的所有客戶端(第一個客戶端除外，它是一個偵聽套接字) foreach ($clients as $send_sock) { // 如果我們從偵聽套接字或客戶端收到消息，請轉到列表中的下一個 if ($send_sock == $sock || $send_sock == $read_sock) continue; // 將消息寫入客戶端-在消息末尾添加換行符 socket_write($send_sock, $data."\n"); } //foreach } } //foreach } // 關閉偵聽套接字 socket_close($sock); ``` ———————————————— 公共聊天室的例子： ``` <?php class webSocket { /** * 服務端地址 * * @var [type] */ private $address; /** * 服務端綁定的端口號 * * @var [type] */ private $port; /** * 監聽端口 * * @var [type] */ private $socket; public function __construct($address='127.0.0.1',$port=9501) { $socket=socket_create(AF_INET,SOCK_STREAM,SOL_TCP); socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1); socket_bind($socket,$address,$port); socket_listen($socket); $this->socket=$socket; $this->port=$port; $this->address=$address; echo mb_convert_encoding('已啟動websockt服務器地址:'.$address.';端口:'.$port.PHP_EOL,'gbk'); } public function run() { //將原客戶端加入監聽連接池 $sockets[]=$this->socket; //定義寫入監聽連接池 $write=null; //定義權限接受連接池 $except=null; //定義超時時間 $time_out=null; //啟動循環阻塞任務 while(true) { //復制連接池 $changes=$sockets; //設置同步阻塞監聽函數 socket_select($changes,$write,$except,$time_out); //監聽端口可讀后操作 foreach ($changes as $sock) { //如果監聽到的是原端口 if($sock==$this->socket) { //讀取錯誤 if(($client=socket_accept($sock))===false) { die('failed to accept socket: '.socket_strerror($sock)."\n"); } //獲取客戶端發送內容 $content=trim(socket_read($client,1024)); //執行http協議升級websocket $this->handshaking($client,$content); $id=rand(1000,9999); $message='id:'.$id.'客戶端已加入連接池'; //通知全體客戶端 $this->sendAll($sockets,$sock,$message); //客戶端加入連接池 $sockets[$id]=$client; //服務端提示 echo mb_convert_encoding('id:'.$id.'客戶端已加入連接池'.PHP_EOL,'gbk'); } //已經握手完畢 else { //接收數據 socket_recv($sock,$buf,1024,0); //群發消息 $str=''; foreach ($sockets as $k => $v) { if($sock==$v) { $str='來自 '.$k.' 的消息:'; break; } } $this->sendAll($sockets,$sock,$str.$this->message($buf)); } } } } /** * 升級協議 * * @param [type] $client * @param [type] $content * @return void */ public function handshaking($client,$content) { //定義頭部信息 $headers=array(); if(preg_match('/Sec-WebSocket-Key:.*\r\n/',$content,$matchs)) { $headers['Sec-WebSocket-Key'] =trim(chop(str_replace('Sec-WebSocket-Key:',"",$matchs[0]))); } //設置返回頭 $secKey = $headers['Sec-WebSocket-Key']; $websocket_accept=base64_encode(pack('H*', sha1($secKey . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'))); $upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" . "Upgrade: websocket\r\n" . "Connection: Upgrade\r\n" . "WebSocket-Origin: $this->address\r\n" . "WebSocket-Location: ws://$this->address:$this->port/websocket/websocket\r\n". "Sec-WebSocket-Accept:$websocket_accept\r\n\r\n"; //寫入緩沖 return socket_write($client,$upgrade,strlen($upgrade)); } /** * 解析接收數據 * @param $buffer * @return null|string */ public function message($buffer) { $len = $masks = $data = $decoded = null; $len = ord($buffer[1]) & 127; if ($len === 126) { $masks = substr($buffer, 4, 4); $data = substr($buffer, 8); } else if ($len === 127) { $masks = substr($buffer, 10, 4); $data = substr($buffer, 14); } else { $masks = substr($buffer, 2, 4); $data = substr($buffer, 6); } for ($index = 0; $index < strlen($data); $index++) { $decoded .= $data[$index] ^ $masks[$index % 4]; } return $decoded; } /** * 消息廣播 * * @param [type] $socket * @param [type] $sock * @param [type] $message * @return void */ public function sendAll($socket,$sock,$message) { foreach ($socket as $so) { if($so!=$sock && $so!=$this->socket) { $this->send($so,$message); } } } /** * 發送數據 * @param $newClinet 新接入的socket * @param $msg 要發送的數據 * @return int|string */ public function send($clinet, $msg){ $msg = $this->frame($msg); socket_write($clinet, $msg, strlen($msg)); } /** * 處理數據幀 * * @param [type] $s * @return void */ public function frame($s) { $a = str_split($s, 125); if (count($a) == 1) { return "\x81" . chr(strlen($a[0])) . $a[0]; } $ns = ""; foreach ($a as $o) { $ns .= "\x81" . chr(strlen($o)) . $o; } return $ns; } } $socketobj=new webSocket(); $socketobj->run(); ``` 前端： ``` <!doctype html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width,initial-scale=1, maximum-scale=1, user-scalable=no"> <title>websocket</title> </head> <body> <input id="text" value=""> <input type="submit" value="send" onclick="start()"> <div id="msg"></div> <script> /** *0：未連接 *1：連接成功，可通訊 *2：正在關閉 *3：連接已關閉或無法打開 */ //創建一個webSocket 實例 var webSocket = new WebSocket("ws://127.0.0.1:9501"); webSocket.onerror = function (event){ onError(event); }; // 打開websocket webSocket.onopen = function (event){ onOpen(event); }; //監聽消息 webSocket.onmessage = function (event){ onMessage(event); }; webSocket.onclose = function (event){ onClose(event); } //關閉監聽websocket function onError(event){ document.getElementById("msg").innerHTML = "<p>close</p>"; console.log("error"+event.data); }; function onOpen(event){ console.log("open:"+sockState()); document.getElementById("msg").innerHTML = "<p>Connect to Service</p>"; }; function onMessage(event){ console.log("onMessage"); document.getElementById("msg").innerHTML += "<p>response:"+event.data+"</p>" }; function onClose(event){ document.getElementById("msg").innerHTML = "<p>close</p>"; console.log("close:"+sockState()); webSocket.close(); } function sockState(){ var status = ['未連接','連接成功，可通訊','正在關閉','連接已關閉或無法打開']; return status[webSocket.readyState]; } function start(event){ console.log(webSocket); var msg = document.getElementById('text').value; document.getElementById('text').value = ''; console.log("send:"+sockState()); console.log("msg="+msg); webSocket.send("msg="+msg); document.getElementById("msg").innerHTML += "<p>準備請求的數據："+msg+"</p>" }; function close(event){ webSocket.close(); } </script> </body> </html> ```