swoole_event_add實現異步

時間 2019-11-10

標籤 swoole event add 實現異步简体版

原文原文鏈接

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

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

異步tcp客戶端

stream_socket_client實現tcp同步客戶端

示例：web

<?php

$start_time = microtime(TRUE);

$fp = stream_socket_client("tcp://www.52fhy.com:80", $errno, $errstr, 30);
fwrite($fp,"GET /test.json HTTP/1.1\r\nHost: www.52fhy.com\r\n\r\n");

echo $resp = fread($fp, 8192);
fclose($fp);
echo "Finish\n";  

$end_time = microtime(TRUE);
echo sprintf("use time:%.3f s\n", $end_time - $start_time);

上述代碼是同步執行的。如何變成異步呢？算法

stream_socket_client實現tcp異步客戶端

因爲fread讀取響應數據是同步堵塞的，咱們將$fp加入到事件監聽後，底層會自動將該socket設置爲非阻塞模式。修改fread那一行：apache

swoole_event_add($fp, function($fp) {
    echo $resp = fread($fp, 8192);
    swoole_event_del($fp);//socket處理完成後，從epoll事件中移除socket
    fclose($fp);
});
echo "Finish\n";  //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函數原型：json

bool swoole_event_add(mixed $sock, mixed $read_callback, mixed $write_callback = null,
    int $flags = null);

$sock能夠爲如下四種類型：segmentfault

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能實時收到，反之也能夠。bash

首先，咱們得建立個tcp_server:websocket

swoole_tcp_server.php

<?php 

$serv = new swoole_server('0.0.0.0', 9001);

$serv->on('Start', function(){
    echo "Tcp server start. Waiting client... \n";
});

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

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

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

$serv->start();

而後實現客戶端：

event_add_tcp_client.php

<?php

$socket = @stream_socket_client("tcp://127.0.0.1:9001", $errno, $errstr, 30);
if(!$socket) exit("connect server err!");

function onRead($socket){
    $msg = stream_socket_recvfrom($socket, 1024);
    if(!$msg){
        swoole_event_del($socket);
    }
    echo "Recv: {$msg}\n";
    fwrite(STDOUT, "ENTER MSG:");
}

function onWrite($socket){
    echo "onWrite\n";
}

function onStdin(){
    global $socket;
    $msg = trim(fgets(STDIN));
    if($msg == 'exit'){ //必須trim此處纔會相等
        swoole_event_exit();
        // exit();
    }
    fwrite($socket, $msg);//數據量大的時候用swoole_event_write
    fwrite(STDOUT, "ENTER MSG:");
}

swoole_event_add($socket, 'onRead', 'onWrite');
swoole_event_add(STDIN, 'onStdin');

fwrite(STDOUT, "ENTER MSG:");  //swoole_event_add不會阻塞進程，這行代碼會順序執行

先運行服務端：

$ 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_*系列函數：

<?php 

$client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
$client->on("connect", function(swoole_client $cli) {
    
});
$client->on("receive", function(swoole_client $cli, $data){
    echo "Receive: $data";
    $cli->send(str_repeat('A', 100)."\n");
    sleep(1);
});
$client->on("error", function(swoole_client $cli){
    echo "error\n";
});
$client->on("close", function(swoole_client $cli){
    echo "Connection close\n";
});
$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}\n");
}
$client->send("hello world\n");
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

<?php

class HttpClient{

    private $callback = null;

    public function __construct($url, $method = 'GET', $postfields = NULL, $headers = array()){
        
        //子進程發起請求
        $process = new swoole_process(function(swoole_process $worker) use($url, $method, $postfields, $headers){
            $response = $this->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 '<pre>';
        //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\n";

運行：

$ 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\n";

輸出：

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) . "\n";
    echo $cli->body;
});

echo "ok\n";

websocket:

<?php 
$cli = new swoole_http_client('118.25.40.163', 8088);

$cli->on('message', function ($_cli, $frame) {
    // var_dump($frame);
    echo $frame->data;
});

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

echo "ok\n";

發送完客戶端會當即close。

參考

一、swoole_event_add https://wiki.swoole.com/wiki/page/119.html 二、Client https://wiki.swoole.com/wiki/page/p-client.html 三、利用swoole_process和eventloop實現php異步編程 https://segmentfault.com/a/1190000008034626 四、關於swoole_process的一些使用疑惑 https://group.swoole.com/question/106198 五、swoole多進程操做 https://blog.csdn.net/koastal/article/details/52871316 六、swoole教程第一節:進程管理模塊(Process)-中(消息隊列) https://segmentfault.com/a/1190000003073389 七、PHP編程中嘗試程序併發的幾種方式總結 http://www.jb51.net/article/81245.htm 八、1.8.0 使用內置Http異步客戶端 https://wiki.swoole.com/wiki/page/678.html 九、異步Http/WebSocket客戶端 https://wiki.swoole.com/wiki/page/p-http_client.html

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。