swoole_event_add實現異步

Cympros 發布于2019-06-28 18:46 / 678人閱讀

摘要：提供了函數，可以實現異步。如何變成異步呢實現異步客戶端由于讀取響應數據是同步堵塞的，我們將加入到事件監聽后，底層會自動將該設置為非阻塞模式。而是基于實現的異步客戶端，沒有并發限制，可在一個進程內同時并發上萬請求。

swoole提供了swoole_event_add函數，可以實現異步。此函數可以用在Server或Client模式下。

swoole_event_add屬于AsyncIO，必須運行在CLI 模式。

異步tcp客戶端 stream_socket_client實現tcp同步客戶端

示例：


上述代碼是同步執行的。如何變成異步呢？
stream_socket_client實現tcp異步客戶端
由于fread讀取響應數據是同步堵塞的，我們將$fp加入到事件監聽后，底層會自動將該socket設置為非阻塞模式。修改fread那一行：
swoole_event_add($fp, function($fp) {
    echo $resp = fread($fp, 8192);
    swoole_event_del($fp);//socket處理完成后，從epoll事件中移除socket
    fclose($fp);
});
echo "Finish
";  //swoole_event_add不會阻塞進程，這行代碼會順序執行
執行后輸出：
Finish
use time:0.087 s
HTTP/1.1 200 OK
Server: AliyunOSS
Date: Sat, 21 Apr 2018 08:36:40 GMT
Content-Type: application/json
Content-Length: 26
Connection: keep-alive
x-oss-request-id: 5ADAF81884D23C965A5D2614
Accept-Ranges: bytes
ETag: "3B3B50D9C802324BB72A74FCD9060004"
Last-Modified: Sat, 21 Apr 2018 04:43:33 GMT
x-oss-object-type: Normal
x-oss-hash-crc64ecma: 9917578698767912878
x-oss-storage-class: Standard
Content-MD5: OztQ2cgCMku3KnT82QYABA==
x-oss-server-time: 5

{"url":"http://52fhy.com"}
swoole_event_add函數原型：
bool swoole_event_add(mixed $sock, mixed $read_callback, mixed $write_callback = null,
    int $flags = null);
$sock可以為以下四種類型：

int，就是文件描述符，包括swoole_client->$sock、swoole_process->$pipe或者其他fd
stream資源，就是stream_socket_client/fsockopen創建的資源
sockets資源，就是sockets擴展中socket_create創建的資源，需要在編譯時加入 ./configure --enable-sockets
object，swoole_process或swoole_client，底層自動轉換為管道或客戶端連接的socket

多個tcp客戶端實時交互
上面的例子，已經實現了異步tcp客戶端。接下來的例子會復雜些：可以在客戶端A輸入，客戶端B能實時收到，反之也可以。
首先，我們得創建個tcp_server:  
swoole_tcp_server.php
on("Start", function(){
    echo "Tcp server start. Waiting client... 
";
});

$serv->on("Connect", function($serv, $fd){
    echo "New client fd:{$fd}. 
";
});

$serv->on("Receive", function($serv, $fd, $from_id, $data){
    echo "Recv msg from fd:{$fd}:{$data}
";
    foreach ($serv->connections as $client) {
        if($fd != $client){
            $serv->send($client, $data);
        }
    }
});

$serv->on("Close", function($serv, $fd){
    echo "Client fd:{$fd} closed. 
";
});

$serv->start();
然后實現客戶端：
event_add_tcp_client.php

先運行服務端：
$ php swoole_tcp_server.php
Tcp server start. Waiting client...
打開兩個終端，運行客戶端：
$ php event_add_tcp_client.php
ENTER MSG:
效果圖：

swoole_client
其實swoole已經提供了異步的swoole_client，無需使用stream_socket_*系列函數：
on("connect", function(swoole_client $cli) {
    
});
$client->on("receive", function(swoole_client $cli, $data){
    echo "Receive: $data";
    $cli->send(str_repeat("A", 100)."
");
    sleep(1);
});
$client->on("error", function(swoole_client $cli){
    echo "error
";
});
$client->on("close", function(swoole_client $cli){
    echo "Connection close
";
});
$client->connect("127.0.0.1", 9001);
還有swoole實現的tcp/udp同步阻塞客戶端：
$client = new swoole_client(SWOOLE_SOCK_TCP);
if (!$client->connect("127.0.0.1", 9001, -1))
{
    exit("connect failed. Error: {$client->errCode}
");
}
$client->send("hello world
");
echo $client->recv();
$client->close();
swoole_client 函數原型：
swoole_client->__construct(int $sock_type, int $is_sync = SWOOLE_SOCK_SYNC, string $key);
可以使用swoole提供的宏來之指定類型，請參考 swoole常量定義


$sock_type表示socket的類型，如TCP/UDP
使用$sock_type | SWOOLE_SSL可以啟用SSL加密
$is_sync表示同步阻塞還是異步非阻塞，默認為同步阻塞

$key用于長連接的Key，默認使用IP:PORT作為key。相同key的連接會被復用

php-fpm/apache環境下只能使用同步客戶端。異步客戶端只能使用在cli命令行環境。
異步http客戶端
curl或者file_get_contents發送http請求是同步阻塞的。基于swoole_event_add封裝可以實現異步。
swoole_event_add實現異步http客戶端
event_add_http_client.php
http($url, $method, $postfields, $headers);
            $worker->write($response);
        }, true);
        $process->start();

        //異步讀取
        swoole_event_add($process->pipe, function($pipe) use ($process){
            $response = $process->read();
            // print_r($response);
            if(is_callable($this->callback)){
                call_user_func($this->callback, $response); //回調
            }
            swoole_event_del($pipe);
        });
    }

    public function setCallback($callback){
        $this->callback = $callback;
    }

    /**
     * http請求
     */
    private function http($url, $method, $postfields = NULL, $headers = array()) {
        try{
            $ssl = stripos($url,"https://") === 0 ? true : false;
            $ci = curl_init();
            /* Curl settings */
            curl_setopt($ci, CURLOPT_USERAGENT, @$_SERVER["HTTP_USER_AGENT"]); //在HTTP請求中包含一個"User-Agent: "頭的字符串。    
            curl_setopt($ci, CURLOPT_CONNECTTIMEOUT, 30);
            curl_setopt($ci, CURLOPT_TIMEOUT, 30);
            curl_setopt($ci, CURLOPT_RETURNTRANSFER, TRUE);
            curl_setopt($ci, CURLOPT_ENCODING, "");
            if ($ssl) {
                curl_setopt($ci, CURLOPT_SSL_VERIFYPEER, 0); // 對認證證書來源的檢查 
                curl_setopt($ci, CURLOPT_SSL_VERIFYHOST, 2); // 從證書中檢查SSL加密算法是否存在
            }
            curl_setopt($ci, CURLOPT_HEADER, FALSE);
    
            switch ($method) {
                case "POST":
                    curl_setopt($ci, CURLOPT_POST, TRUE);
                    if (!empty($postfields)) {
                        curl_setopt($ci, CURLOPT_POSTFIELDS, $postfields);
                    }
                    break;
            }
    
            curl_setopt($ci, CURLOPT_URL, $url );
            curl_setopt($ci, CURLOPT_HTTPHEADER, $headers );
            curl_setopt($ci, CURLINFO_HEADER_OUT, TRUE );
    
            $response = curl_exec($ci);
            $httpCode = curl_getinfo($ci, CURLINFO_HTTP_CODE);
            $httpInfo = curl_getinfo($ci);
            
            if (FALSE === $response)
                throw new Exception(curl_error($ci), curl_errno($ci));
        
        } catch(Exception $e) {
            throw $e;
        }
        
        //echo "";
        //var_dump($response);
        //var_dump($httpInfo);

        curl_close ($ci);
        return $response;
    }
}

$client = new HttpClient("http://www.52fhy.com/test.json");
$client->setCallback(function($response){
    print_r($response);
});
echo "OK
";


運行：
$ php event_add_http_client.php
OK
{"url":"http://52fhy.com"}[
由返回結果可以看出，客戶端請求是異步執行的。
swoole_http_client
Swoole也內置了http異步客戶端（swoole>=1.8.0）。
相比curl和file_get_contents這樣PHP提供的Http客戶端，swoole_http_client最大的優勢是支持大量并發。
file_get_contents只能同時請求一個URL，并發只能通過開啟多進程實現。curl提供了curl_multi功能實現并發基于select和多線程。并發能力都很差。而swoole_http_client是基于epoll實現的異步客戶端，沒有并發限制，可在一個進程內同時并發上萬請求。更多介紹詳見swoole文檔。
示例：
get:
$cli = new swoole_http_client("www.52fhy.com", 80);
$cli->setHeaders(["User-Agent" => "swoole"]);

$cli->get("/test.json", function ($cli) {
    echo $cli->body;
});

echo "ok
";
輸出：
ok
{"url":"http://52fhy.com"}
post:
$cli = new swoole_http_client("127.0.0.1", 81); 
$cli->post("/post_demo.php", array("a" => "1234", "b" => "456"), function ($cli) {
    echo "Length: " . strlen($cli->body) . "
";
    echo $cli->body;
});

echo "ok
";
websocket:
on("message", function ($_cli, $frame) {
    // var_dump($frame);
    echo $frame->data;
});

$cli->upgrade("/", function ($cli) {
    $cli->push("hello world");
});

echo "ok
";
發送完客戶端會立即close。
參考
1、swoole_event_add 
https://wiki.swoole.com/wiki/...
2、Client 
https://wiki.swoole.com/wiki/...
3、利用swoole_process和eventloop實現php異步編程
https://segmentfault.com/a/11...
4、關于swoole_process的一些使用疑惑
https://group.swoole.com/ques...
5、swoole多進程操作
https://blog.csdn.net/koastal...
6、swoole教程第一節:進程管理模塊(Process)-中(消息隊列)
https://segmentfault.com/a/11...
7、PHP編程中嘗試程序并發的幾種方式總結
http://www.jb51.net/article/8...
8、1.8.0 使用內置Http異步客戶端
https://wiki.swoole.com/wiki/...
9、異步Http/WebSocket客戶端
https://wiki.swoole.com/wiki/...

GPU云服務器云服務器實現異步調用 swoole_event_add 接口異步異步采集

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/28639.html

swoole_process父子進程管道通信案例

摘要：話不多說直接上代碼創建的子進程獲取異步獲取更高性能啟動子進程子進程處理邏輯異步非阻塞網關連接失敗讀取父進程管道消息父進程獲取子進程的管道消息子進程消息子進程的客戶端可以忽略不計，本只是話不多說直接上代碼創建的子進程： public function __construct() { $this->redis = Container::get(SwooleR...

Cheng_Gang 2019-07-01 12:32 評論0 收藏0
Swoole 源碼分析——Async 異步事件系統 Swoole_Event

摘要：利用將傳入的轉為文件描述符新建對象，并對其設置文件描述符讀寫回調函數檢測是否存在，并對其進行初始化。如果傳入在中不存在返回，用于修改事件監聽的回調函數和掩碼。異常事件回調函數當發現套接字發生錯誤后，就會自動刪除該套接字的監聽。前言對于異步的任務來說，Server 端的 master 進程與 worker 進程會自動將異步的事件添加到 reactor 的事件循環中去，task_wor...

stefanieliang 2019-07-01 10:45 評論0 收藏0
基于Swoole和Redis實現的并發隊列處理系統

摘要：大家知道，一個消息隊列處理系統主要分為兩大部分消費者和生產者。任務系統實時的對任務隊列進行，出來一個任務就一個子進程，由子進程完成具體的任務邏輯。新的設計為了解決并發的問題，我們計劃做一個更加高效強壯的隊里處理系統。背景由于PHP不支持多線程，但是作為一個完善的系統，有很多操作都是需要異步完成的。為了完成這些異步操作，我們做了一個基于Redis隊列任務系統。大家知道，一個消息隊列...

booster 2019-07-01 12:33 評論0 收藏0
使用 swoole_process 實現 PHP 進程池

摘要：本文使用與完成一個的進程池，并且支持動態創建新進程。接著遍歷所有的進程，并且加入中，設置可讀事件，用于接收子進程的空閑信號。最后每隔一秒向進程投遞任務。由于只模擬了十次任務，則第十個任務完成之后在父進程中發送使所有子進程退出。 swoole_process 主要是用來代替 PHP 的 pcntl 擴展。我們知道 pcntl 是用來進行多進程編程的，而 pcntl 只提供了 fork 這...

Andrman 2019-07-01 10:52 評論0 收藏0