AsyncTcpConnection類 · PHP筆記

``` 讓Workerman作為客戶端向遠程服務端發起異步連接，并通過send接口和onMessage回調異步發送和處理連接上的數據。 ``` >[danger]AsyncTcpConnection繼承至TcpConnection，所以TcpConnection的方法合屬性也可以用在AsyncTcpConnection中使用目前AsyncTcpConnection支持的協議有tcp、ssl、ws、frame、text、自定義協議。目前不支持http協議其中ssl要求Workerman>=3.3.4，并安裝openssl擴展 [TOC] 常量： ``` { "READ_BUFFER_SIZE": 65535, "STATUS_INITIAL": 0, "STATUS_CONNECTING": 1, "STATUS_ESTABLISHED": 2, "STATUS_CLOSING": 4, "STATUS_CLOSED": 8 } ``` 屬性： json_encode(get_object_vars($connection), JSON_UNESCAPED_UNICODE); ``` { "onMessage": { }, "onClose": { }, "onError": { }, "onBufferFull": { }, "onBufferDrain": { }, "protocol": null, "transport": "tcp", "worker": null, "bytesRead": 0, "bytesWritten": 0, "id": 1, "maxSendBufferSize": 1024, "context": { }, "maxPackageSize": 10485760, "onConnect": { } } ``` json_encode(get_class_vars(get_class($connection)), JSON_UNESCAPED_UNICODE); ``` { "onConnect": null, "transport": "tcp", "onMessage": null, "onClose": null, "onError": null, "onBufferFull": null, "onBufferDrain": null, "protocol": null, "worker": null, "bytesRead": 0, "bytesWritten": 0, "id": 0, "maxSendBufferSize": 1048576, "context": null, "maxPackageSize": 1048576, "defaultMaxSendBufferSize": 1048576, "defaultMaxPackageSize": 10485760, "connections": [ ], "_statusToString": { "0": "INITIAL", "1": "CONNECTING", "2": "ESTABLISHED", "4": "CLOSING", "8": "CLOSED" }, "statistics": { "connection_count": 0, "total_request": 0, "throw_exception": 0, "send_fail": 0 } } ``` json_encode($r->getProperties(\ReflectionMethod::IS_PUBLIC), JSON_UNESCAPED_UNICODE); ``` [ { "name": "onConnect", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "transport", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "onMessage", "class": "Workerman\Connection\TcpConnection" }, { "name": "onClose", "class": "Workerman\Connection\TcpConnection" }, { "name": "onError", "class": "Workerman\Connection\TcpConnection" }, { "name": "onBufferFull", "class": "Workerman\Connection\TcpConnection" }, { "name": "onBufferDrain", "class": "Workerman\Connection\TcpConnection" }, { "name": "protocol", "class": "Workerman\Connection\TcpConnection" }, { "name": "worker", "class": "Workerman\Connection\TcpConnection" }, { "name": "bytesRead", "class": "Workerman\Connection\TcpConnection" }, { "name": "bytesWritten", "class": "Workerman\Connection\TcpConnection" }, { "name": "id", "class": "Workerman\Connection\TcpConnection" }, { "name": "maxSendBufferSize", "class": "Workerman\Connection\TcpConnection" }, { "name": "context", "class": "Workerman\Connection\TcpConnection" }, { "name": "defaultMaxSendBufferSize", "class": "Workerman\Connection\TcpConnection" }, { "name": "maxPackageSize", "class": "Workerman\Connection\TcpConnection" }, { "name": "defaultMaxPackageSize", "class": "Workerman\Connection\TcpConnection" }, { "name": "connections", "class": "Workerman\Connection\TcpConnection" }, { "name": "_statusToString", "class": "Workerman\Connection\TcpConnection" }, { "name": "statistics", "class": "Workerman\Connection\ConnectionInterface" } ] ``` 方法 json_encode(get_class_methods($connection), JSON_UNESCAPED_UNICODE); ``` [ "__construct", "connect", "reconnect", "cancelReconnect", "getRemoteHost", "getRemoteURI", "checkConnection", "getStatus", "send", "getRemoteIp", "getRemotePort", "getRemoteAddress", "getLocalIp", "getLocalPort", "getLocalAddress", "getSendBufferQueueSize", "getRecvBufferQueueSize", "isIpV4", "isIpV6", "pauseRecv", "resumeRecv", "baseRead", "baseWrite", "doSslHandshake", "pipe", "consumeRecvBuffer", "close", "getSocket", "bufferIsEmpty", "destroy", "__destruct" ] ``` //$r = new \ReflectionObject($connection); json_encode($r->getMethods(\ReflectionMethod::IS_PUBLIC), JSON_UNESCAPED_UNICODE); ``` [ { "name": "__construct", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "connect", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "reconnect", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "cancelReconnect", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "getRemoteHost", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "getRemoteURI", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "checkConnection", "class": "Workerman\Connection\AsyncTcpConnection" }, { "name": "getStatus", "class": "Workerman\Connection\TcpConnection" }, { "name": "send", "class": "Workerman\Connection\TcpConnection" }, { "name": "getRemoteIp", "class": "Workerman\Connection\TcpConnection" }, { "name": "getRemotePort", "class": "Workerman\Connection\TcpConnection" }, { "name": "getRemoteAddress", "class": "Workerman\Connection\TcpConnection" }, { "name": "getLocalIp", "class": "Workerman\Connection\TcpConnection" }, { "name": "getLocalPort", "class": "Workerman\Connection\TcpConnection" }, { "name": "getLocalAddress", "class": "Workerman\Connection\TcpConnection" }, { "name": "getSendBufferQueueSize", "class": "Workerman\Connection\TcpConnection" }, { "name": "getRecvBufferQueueSize", "class": "Workerman\Connection\TcpConnection" }, { "name": "isIpV4", "class": "Workerman\Connection\TcpConnection" }, { "name": "isIpV6", "class": "Workerman\Connection\TcpConnection" }, { "name": "pauseRecv", "class": "Workerman\Connection\TcpConnection" }, { "name": "resumeRecv", "class": "Workerman\Connection\TcpConnection" }, { "name": "baseRead", "class": "Workerman\Connection\TcpConnection" }, { "name": "baseWrite", "class": "Workerman\Connection\TcpConnection" }, { "name": "doSslHandshake", "class": "Workerman\Connection\TcpConnection" }, { "name": "pipe", "class": "Workerman\Connection\TcpConnection" }, { "name": "consumeRecvBuffer", "class": "Workerman\Connection\TcpConnection" }, { "name": "close", "class": "Workerman\Connection\TcpConnection" }, { "name": "getSocket", "class": "Workerman\Connection\TcpConnection" }, { "name": "bufferIsEmpty", "class": "Workerman\Connection\TcpConnection" }, { "name": "destroy", "class": "Workerman\Connection\TcpConnection" }, { "name": "__destruct", "class": "Workerman\Connection\TcpConnection" } ] ``` ## **異步訪問外部http服務** ``` require_once __DIR__ . '/vendor/autoload.php'; use Workerman\Worker; use Workerman\Connection\TcpConnection; use Workerman\Connection\AsyncTcpConnection; $task = new Worker(); // 進程啟動時異步建立一個到www.baidu.com連接對象，并發送數據獲取數據 $task->onWorkerStart = function($task){ // 不支持直接指定http，但是可以用tcp模擬http協議發送數據 $connection_to_baidu = new AsyncTcpConnection('tcp://www.baidu.com:80'); // 當連接建立成功時，發送http請求數據 $connection_to_baidu->onConnect = function(AsyncTcpConnection $connection_to_baidu){ echo "connect success\n"; $connection_to_baidu->send("GET / HTTP/1.1\r\nHost: www.baidu.com\r\nConnection: keep-alive\r\n\r\n"); }; $connection_to_baidu->onMessage = function(AsyncTcpConnection $connection_to_baidu, $http_buffer){ echo $http_buffer; }; $connection_to_baidu->onClose = function(AsyncTcpConnection $connection_to_baidu){ echo "connection closed\n"; }; $connection_to_baidu->onError = function(AsyncTcpConnection $connection_to_baidu, $code, $msg){ echo "Error code:$code msg:$msg\n"; }; $connection_to_baidu->connect(); }; Worker::runAll(); ``` ## **示例 2、異步訪問外部websocket服務，并設置以哪個本地ip及端口訪問** ``` require_once __DIR__ . '/vendor/autoload.php'; use Workerman\Worker; use Workerman\Connection\TcpConnection; use Workerman\Connection\AsyncTcpConnection; $worker = new Worker(); $worker->onWorkerStart = function($worker) { // 如果你想只在一個進程里發起連接，可以通過判斷$worker->id來做到，例如下面是只在0號進程發起連接 if ($worker->id != 0) { return; } // 設置訪問對方主機的本地ip及端口(每個socket連接都會占用一個本地端口) $context_option = array( 'socket' => array( // ip必須是本機網卡ip，并且能訪問對方主機，否則無效 'bindto' => '114.215.84.87:2333', ), ); $con = new AsyncTcpConnection('ws://echo.websocket.org:80', $context_option); $con->onConnect = function(AsyncTcpConnection $con) { $con->send('hello'); }; $con->onMessage = function(AsyncTcpConnection $con, $data) { echo $data; }; $con->connect(); }; Worker::runAll(); ``` 示例 3、異步訪問外部wss端口，并設置本地ssl證書 ``` require_once __DIR__ . '/vendor/autoload.php'; use Workerman\Worker; use Workerman\Connection\TcpConnection; use Workerman\Connection\AsyncTcpConnection; $worker = new Worker(); $worker->onWorkerStart = function($worker){ // 設置訪問對方主機的本地ip及端口以及ssl證書 $context_option = array( 'socket' => array( // ip必須是本機網卡ip，并且能訪問對方主機，否則無效 'bindto' => '114.215.84.87:2333', ), // ssl選項，參考https://php.net/manual/zh/context.ssl.php 'ssl' => array( // 本地證書路徑。必須是 PEM 格式，并且包含本地的證書及私鑰。 'local_cert' => '/your/path/to/pemfile', // local_cert 文件的密碼。 'passphrase' => 'your_pem_passphrase', // 是否允許自簽名證書。 'allow_self_signed' => true, // 是否需要驗證 SSL 證書。 'verify_peer' => false ) ); // 發起異步連接 $con = new AsyncTcpConnection('ws://echo.websocket.org:443', $context_option); // 設置以ssl加密方式訪問 $con->transport = 'ssl'; $con->onConnect = function(AsyncTcpConnection $con) { $con->send('hello'); }; $con->onMessage = function(AsyncTcpConnection $con, $data) { echo $data; }; $con->connect(); }; Worker::runAll(); ``` ## **Mysql代理** `mysql -uroot -P4406 -h127.0.0.1 -p` 代理后后臺鏈接數據庫時將3306換成4406也能成功登陸了 ``` require_once __DIR__ . '/vendor/autoload.php'; use Workerman\Worker; use Workerman\Connection\TcpConnection; use Workerman\Connection\AsyncTcpConnection; // 真實的mysql地址，假設這里是本機3306端口 $REAL_MYSQL_ADDRESS = 'tcp://127.0.0.1:3306'; // 代理監聽本地4406端口 $proxy = new Worker('tcp://0.0.0.0:4406'); $proxy->onConnect = function(TcpConnection $connection){ global $REAL_MYSQL_ADDRESS; // 異步建立一個到實際mysql服務器的連接 $connection_to_mysql = new AsyncTcpConnection($REAL_MYSQL_ADDRESS); // mysql連接發來數據時，轉發給對應客戶端的連接 $connection_to_mysql->onMessage = function(AsyncTcpConnection $connection_to_mysql, $buffer)use($connection){ $connection->send($buffer); }; // mysql連接關閉時，關閉對應的代理到客戶端的連接 $connection_to_mysql->onClose = function(AsyncTcpConnection $connection_to_mysql)use($connection){ $connection->close(); }; // mysql連接上發生錯誤時，關閉對應的代理到客戶端的連接 $connection_to_mysql->onError = function(AsyncTcpConnection $connection_to_mysql)use($connection){ $connection->close(); }; // 執行異步連接 $connection_to_mysql->connect(); // 客戶端發來數據時，轉發給對應的mysql連接 $connection->onMessage = function(TcpConnection $connection, $buffer)use($connection_to_mysql) { $connection_to_mysql->send($buffer); }; // 客戶端連接斷開時，斷開對應的mysql連接 $connection->onClose = function(TcpConnection $connection)use($connection_to_mysql) { $connection_to_mysql->close(); }; // 客戶端連接發生錯誤時，斷開對應的mysql連接 $connection->onError = function(TcpConnection $connection)use($connection_to_mysql) { $connection_to_mysql->close(); }; }; // 運行worker Worker::runAll(); ``` ## **reConnect重新連接** ~~~php use Workerman\Worker; use Workerman\Connection\AsyncTcpConnection; use Workerman\Connection\TcpConnection; require_once __DIR__ . '/vendor/autoload.php'; $worker = new Worker(); $worker->onWorkerStart = function($worker){ $con = new AsyncTcpConnection('ws://echo.websocket.org:80'); $con->onConnect = function(AsyncTcpConnection $con) { $con->send('hello'); }; $con->onMessage = function(AsyncTcpConnection $con, $msg) { echo "recv $msg\n"; }; $con->onClose = function(AsyncTcpConnection $con) { // 如果連接斷開，則在1秒后重連 $con->reConnect(1); }; $con->connect(); }; Worker::runAll(); ~~~ 調用reconnect成功重連后，$con的onConnect方法會再次被調用(如果有設置的話)。有時候我們想讓onConnect方法只執行一次，重連的時候不再執行，參考如下例子： ~~~php use Workerman\Worker; use Workerman\Connection\AsyncTcpConnection; use Workerman\Connection\TcpConnection; require_once __DIR__ . '/vendor/autoload.php'; $worker = new Worker(); $worker->onWorkerStart = function($worker){ $con = new AsyncTcpConnection('ws://echo.websocket.org:80'); $con->onConnect = function(AsyncTcpConnection $con) { static $is_first_connect = true; if (!$is_first_connect) return; $is_first_connect = false; $con->send('hello'); }; $con->onMessage = function(AsyncTcpConnection $con, $msg) { echo "recv $msg\n"; }; $con->onClose = function(AsyncTcpConnection $con) { // 如果連接斷開，則在1秒后重連 $con->reConnect(1); }; $con->connect(); }; Worker::runAll(); ~~~ ## **transport：設置傳輸屬性，可選值為tcp和ssl，默認是tcp** ``` require_once __DIR__ . '/vendor/autoload.php'; use Workerman\Worker; use Workerman\Connection\TcpConnection; use Workerman\Connection\AsyncTcpConnection; $task = new Worker(); // 進程啟動時異步建立一個到www.baidu.com連接對象，并發送數據獲取數據 $task->onWorkerStart = function($task){ $connection_to_baidu = new AsyncTcpConnection('tcp://www.baidu.com:443'); // 設置為ssl加密連接 $connection_to_baidu->transport = 'ssl'; $connection_to_baidu->onConnect = function(AsyncTcpConnection $connection_to_baidu){ echo "connect success\n"; $connection_to_baidu->send("GET / HTTP/1.1\r\nHost: www.baidu.com\r\nConnection: keep-alive\r\n\r\n"); }; $connection_to_baidu->onMessage = function(AsyncTcpConnection $connection_to_baidu, $http_buffer){ echo $http_buffer; }; $connection_to_baidu->onClose = function(AsyncTcpConnection $connection_to_baidu){ echo "connection closed\n"; }; $connection_to_baidu->onError = function(AsyncTcpConnection $connection_to_baidu, $code, $msg){ echo "Error code:$code msg:$msg\n"; }; $connection_to_baidu->connect(); }; // 運行worker Worker::runAll(); ```