Workerman官方教程學習筆記

視頻教程
 文檔手冊
教程基于 workerman 3.3

基礎教程

Worker類的使用

WorkerMan中有兩個重要的類Worker與Connection。
worker 對象實際上是一個容器，它可以以特定的協議去監聽某個端口。當客戶端連接到這個容器監聽端口之后，會在這個 worker 容器內部產生一個 connection 對象。
在 WorkerMan 中通過操作這個 connection 對象來完成對客戶端發送數據以及接收客戶端數據的操作。

<?php

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Request;

require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('http://0.0.0.0:8686');

$worker->onWorkerStart = function ($worker) {
};
$worker->onConnect = function ($connection) {
    echo "connection success" . PHP_EOL;
};
$worker->onMessage = function (TcpConnection $connection, Request $request) {
    $connection->send("hello");
};
$worker->onClose = function ($connection) {
    echo "connection close" . PHP_EOL;
};
$worker->onWorkerStop = function ($worker) {
};

// 運行worker
Worker::runAll();

<?php

use Workerman\Worker;
use Workerman\Connection\TcpConnection;

require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('websocket://0.0.0.0:8686');

$worker->onMessage = function (TcpConnection $connection, $data) {
    $connection->send("hello");
};

// 運行worker
Worker::runAll();

<?php

use Workerman\Worker;
use Workerman\Connection\TcpConnection;

require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('tcp://0.0.0.0:80');

$worker->onMessage = function (TcpConnection $connection, $data) {
    $connection->send("hello world");
};

// 運行worker
Worker::runAll();

<?php

use Workerman\Worker;
use Workerman\Connection\UdpConnection;

require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('udp://0.0.0.0:80');

$worker->onMessage = function (UdpConnection $connection, $data) {
    $connection->send("hello world");
};

// 運行worker
Worker::runAll();

Connection類的使用

TcpConnection類和AsyncTcpConnection類。AsyncTcpConnection 類是 TcpConnection 類的子類。
當我們在 Workerman 中需要訪問外部的某個服務的時候，通過 AsyncTcpConnection 類異步的發起一個 tcp 連接，去連接遠程的服務端，異步通訊。

<?php

use Workerman\Connection\ConnectionInterface;
use Workerman\Worker;

require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('tcp://0.0.0.0:8686');

$worker->onConnect = function (ConnectionInterface $connection) {
    echo "connection success" . PHP_EOL;
    //類似于IP白名單
    if ($connection->getRemoteIp() !== '127.0.0.1') {
        $connection->close('bad ip');
    }
};
$worker->onMessage = function (ConnectionInterface $connection, $data) {
    $connection->send("hello");
};
$worker->onClose = function ($connection) {
    echo "connection close" . PHP_EOL;
};

// 運行worker
Worker::runAll();

<?php

use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;

require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker('http://0.0.0.0:8686');

$worker->onConnect = function (TcpConnection $connection) {
    // 作為客戶端去鏈接其他服務端（代理）
    // 發起異步的 TCP 連接 通過實現 onConnect、onMessage、onClose 實現業務邏輯
    $connectionBaidu = new AsyncTcpConnection('ssl://www.baidu.com:443');
  	// 當百度返回數據后，轉發給瀏覽器客戶端
    $connectionBaidu->onMessage = function ($connectionBaidu, $data) use ($connection) {
        $connection->send($data);
    };
  	// 瀏覽器客戶端發來數據后轉發給百度
    $connection->onMessage = function ($connection, $data) use ($connectionBaidu) {
        $connectionBaidu->send($data);
    };
  	// 執行連接
    $connectionBaidu->connect();
};

// 運行worker
Worker::runAll();

Timer類的使用

<?php

require_once './vendor/autoload.php';

use Workerman\Worker;
use Workerman\Lib\Timer;

$worker = new Worker('websocket://0.0.0.0:8686');
$worker->onConnect = function ($connection) {
    // 為每個客戶端創建10s的賬戶認證
    Timer::add(10, function () use ($connection) {
        if (!isset($connection->name)) {
            $connection->close('auth timeout and close');
        }
    }, null, false);
};

$worker->onMessage = function ($connection, $data) {
    if (!isset($connection->name)) {
        $data = json_decode($data, true);
        if (!isset($data['name']) || !isset($data['password'])) {
            $connection->close('auth fail and close');
            return;
        }
        //假設此處進行數據庫驗證
        $connection->name = $data['name'];
        broadcast($connection->name . " login");
        return;
    }
    broadcast($connection->name . " said:" . $data);
};

function broadcast($msg)
{
    global $worker;
    foreach ($worker->connections as $connection) {
        if (!isset($connection->name)) {
            continue;
        }
        $connection->send($msg);
    }
}

Worker::runAll();

WebServer的使用

Workerman 4.x 版本去掉了 WebServer，推薦使用 webman。

https://www.workerman.net/q/5072

原理解析

Stream 函數

Worker類

<?php

/**
 * 單進程IO復用select
 * 同時處理多個連接
 * ab -n100000 -c100 -k http://127.0.0.1:1215/
 */
class Worker
{
    // 連接事件回調
    public $onConnect = null;
    // 消息事件回調
    public $onMessage = null;
    // 連接關閉事件回調
    public $onClose = null;
    // 監聽端口的socket
    protected $socket = null;
    // 所有socket，包括客戶端socket和監聽端口socket
    protected $allSockets = [];

    // 構造函數
    public function __construct($address)
    {
        // 創建監聽socket
        $this->socket = stream_socket_server($address, $errno, $errstr);
        echo "listen $address\r\n";
        // 設置為非阻塞
        // 阻塞：當讀取一個socket時，如果對方沒有發來任何數據，這個讀取操作會一直卡著，直到這個socket超時
        // 非阻塞：當讀取一個socket時，如果對方沒有發來任何數據，讀取操作會立刻返回，返回空數據
        // 在workerman中不想因為讀取某個客戶端的連接，而導致整個workerman卡住，所以設置非堵塞
        stream_set_blocking($this->socket, 0);
        // 將監聽socket放入allSockets
        $this->allSockets[(int)$this->socket] = $this->socket;
    }

    // 運行
    public function run()
    {
        while (1) {
            // 這里不監聽socket可寫事件和外帶數據可讀事件
            $write = $except = null;
            // 監聽所有socket可讀事件，包括客戶端socket和監聽端口的socket
            $read = $this->allSockets;
            // stream_select是個IO復用函數，可以監聽一個socket集合的可讀、可寫。
            // 整個程序阻塞在這里，等待$read里面的socket可讀，這里$read是個引用參數，也就是說stream_select()返回后$read是會被重新賦值
            // $read值的內容就是所有狀態可讀的socket集合
            stream_select($read, $write, $except, 60);
            // $read被重新賦值，遍歷所有狀態為可讀的socket
            foreach ($read as $index => $socket) {
                // 如果是監聽socket可讀，說明有新連接
                if ($socket === $this->socket) {
                    // 通過stream_socket_accept獲取新連接（客戶端）
                    $new_conn_socket = stream_socket_accept($this->socket);
                    if (!$new_conn_socket) continue;
                    // 如果有onConnect事件回調，則嘗試觸發
                    if ($this->onConnect) {
                        call_user_func($this->onConnect, $new_conn_socket);
                    }
                    // 將新的客戶端連接socket放入allSockets，以便stream_select監聽其可讀事件
                    $this->allSockets[(int)$new_conn_socket] = $new_conn_socket;
                } else { 
                    // 如果是客戶端連接可讀，說明對應連接的客戶端有數據發來
                    // 讀數據
                    $buffer = fread($socket, 65535);
                    // 數據為空，代表連接已經斷開
                    if ($buffer === '' || $buffer === false) {
                        // 嘗試觸發onClose回調
                        if ($this->onClose) {
                            call_user_func($this->onClose, $socket);
                        }
                        fclose($socket);
                        // 從allSockets里刪除對應的連接，不再監聽這個socket可讀事件
                        unset($this->allSockets[(int)$socket]);
                        continue;
                    }
                    // 嘗試觸發onMessage回調
                    call_user_func($this->onMessage, $socket, $buffer);
                }
            }
        } //end while
    }
}

$server = new Worker('tcp://0.0.0.0:1215');

$server->onConnect = function ($conn) {
    echo "onConnect\n";
};

$server->onMessage = function ($conn, $msg) {
    fwrite($conn, "HTTP/1.1 200 OK\r\nConnection: keep-alive\r\nServer: workerman/1.1.4\r\nContent-length:5\r\n\r\nhello");
};

$server->onClose = function ($conn) {
    echo "onClose\n";
};

$server->run();

TcpConnection類

<?php

/**
 * 單進程IO復用select
 * 同時處理多個連接
 * ab -n100000 -c100 -k http://127.0.0.1:1215/
 */
class Worker
{
    // 連接事件回調
    public $onConnect = null;
    // 消息事件回調
    public $onMessage = null;
    // 連接關閉事件回調
    public $onClose = null;
    // 監聽端口的socket
    protected $socket = null;
    // 所有socket，包括客戶端socket和監聽端口socket
    protected $allSockets = [];
    // 所有tcpconnection類的實例，也就是所有客戶端的連接對象
    public $connections = [];

    // 構造函數
    public function __construct($address)
    {
        // 創建監聽socket
        $this->socket = stream_socket_server($address, $errno, $errstr);
        echo "listen $address\r\n";
        // 設置為非阻塞
        // 阻塞：當讀取一個socket時，如果對方沒有發來任何數據，這個讀取操作會一直卡著，直到這個socket超時
        // 非阻塞：當讀取一個socket時，如果對方沒有發來任何數據，讀取操作會立刻返回，返回空數據
        stream_set_blocking($this->socket, 0);
        // 將監聽socket放入allSockets
        $this->allSockets[(int)$this->socket] = $this->socket;
    }

    // 運行
    public function run()
    {
        while (1) {
            // 這里不監聽socket可寫事件和外帶數據可讀事件
            $write = $except = null;
            // 監聽所有socket可讀事件，包括客戶端socket和監聽端口的socket
            $read = $this->allSockets;
            // stream_select是個IO復用函數。整個程序阻塞在這里，等待$read里面的socket可讀，這里$read是個引用參數
            stream_select($read, $write, $except, 60);
            // $read被重新賦值，遍歷所有狀態為可讀的socket
            foreach ($read as $index => $socket) {
                // 如果是監聽socket可讀，說明有新連接
                if ($socket === $this->socket) {
                    // 通過stream_socket_accept獲取新連接
                    $new_conn_socket = stream_socket_accept($this->socket);
                    if (!$new_conn_socket) continue;
                    $conn = new TcpConnection($new_conn_socket);
                    $this->connections[(int)$new_conn_socket] = $conn;
                    // 如果有onConnect事件回調，則嘗試觸發
                    if ($this->onConnect) {
                        call_user_func($this->onConnect, $conn);
                    }
                    // 將新的客戶端連接socket放入allSockets，以便stream_select監聽其可讀事件
                    $this->allSockets[(int)$new_conn_socket] = $new_conn_socket;
                } else { // 是客戶端連接可讀，說明對應連接的客戶端有數據發來
                    // 讀數據
                    $buffer = fread($socket, 65535);
                    // 數據為空，代表連接已經斷開
                    if ($buffer === '' || $buffer === false) {
                        // 嘗試觸發onClose回調
                        if ($this->onClose) {
                            call_user_func($this->onClose, $this->connections[(int)$socket]);
                        }
                        fclose($socket);
                        // 從allSockets里刪除對應的連接，不再監聽這個socket可讀事件
                        unset($this->allSockets[(int)$socket], $this->connections[(int)$socket]);
                        continue;
                    }
                    // 嘗試觸發onMessage回調
                    call_user_func($this->onMessage, $this->connections[(int)$socket], $buffer);
                }
            }
        } //end while
    }
}

問題：

上述 onConnect、onMessage、onClose 回調中需要使用 php 原始 api 來操作 socket
TcpConnection 其實是對這些 socket 的二次封裝

<?php

class TcpConnection
{
    protected $_socket = null;

    public function __construct($socket)
    {
        $this->_socket = $socket;
    }

    public function send($buffer)
    {
        //檢測socket是否斷開
        if (feof($this->_socket)) return false;
        return fwrite($this->_socket, $buffer);
    }
}

<?php

require_once 'Worker.php';
require_once 'TcpConnection.php';

$server = new Worker('tcp://0.0.0.0:1215');

$server->onConnect = function ($conn) {
    $conn->send('input your name: ');
};

$server->onMessage = function ($conn, $msg) {
    if (!isset($conn->name)) {
        $conn->name = trim($msg);
        broadcast("{$conn->name} come");
        return;
    }
    broadcast("{$conn->name} said: $msg");
};

$server->onClose = function ($conn) {
    broadcast("{$conn->name} logout");
};

function broadcast($msg)
{
    global $server;
    foreach ($server->connections as $conn) {
        $conn->send("$msg\r\n");
    }
}

$server->run();

源碼解析

客戶端與worker進程的關系

連接哪個Worker進程由操作系統根據系統的負載情況自動分配

主進程與worker子進程關系

Workerman                      // workerman內核代碼
    ├── Connection                 // socket連接相關
    │   ├── ConnectionInterface.php// socket連接接口
    │   ├── TcpConnection.php      // Tcp連接類
    │   ├── AsyncTcpConnection.php // 異步Tcp連接類
    │   └── UdpConnection.php      // Udp連接類
    ├── Events                     // 網絡事件庫
    │   ├── EventInterface.php     // 網絡事件庫接口
    │   ├── Event.php              // Libevent網絡事件庫
    │   ├── Ev.php                 // Libev網絡事件庫
    │   ├── Swoole.php             // Swoole網絡事件庫
    │   └── Select.php             // Select網絡事件庫
    ├── Lib                        // 常用的類庫
    │   ├── Constants.php          // 常量定義
    │   └── Timer.php              // 定時器
    ├── Protocols                  // 協議相關
    │   ├── ProtocolInterface.php  // 協議接口類
    │   ├── Http                   // http協議相關
    │   │   ├── Chunk.php    // http chunk類
    │   │   ├── Request.php  // http 請求類
    │   │   ├── Response.php  // http響應類
    │   │   ├── ServerSentEvents.php  // SSE類
    │   │   ├── Session
    │   │   │   ├── FileSessionHandler.php  // session文件存儲
    │   │   │   └── RedisSessionHandler.php // session redis存儲
    │   │   ├── Session.php  // session類
    │   │   └── mime.types   // mime映射文件
    │   ├── Http.php               // http協議實現
    │   ├── Text.php               // Text協議實現
    │   ├── Frame.php              // Frame協議實現
    │   └── Websocket.php          // websocket協議的實現
    ├── Worker.php                 // Worker
    ├── WebServer.php              // WebServer
    └── Autoloader.php             // 自動加載類

事件循環

event
- 這是一個擴展，有效地調度I/O，時間和信號基礎事件使用最好的可用于特定平臺的I/O通知機制。這是將libevent移植到PHP基礎設施的一個端口。
libevent
- Libevent 是一個庫，它提供了一種機制，在文件描述符上發生特定事件時或超時后執行回調函數。
- PHP8 開始在文檔中被移除。。。
select
- stream_select - 在給定的流數組上運行與 select() 系統調用等效的操作，超時由 tv_sec 和 tv_usec 指定

protected static $_availableEventLoops = array(
  'event'    => '\Workerman\Events\Event',
  'libevent' => '\Workerman\Events\Libevent'
);

/**
 * Get event loop name.
 *
 * @return string
 */
protected static function getEventLoopName()
{
    if (static::$eventLoopClass) {
        return static::$eventLoopClass;
    }

    if (!\class_exists('\Swoole\Event', false)) {
        unset(static::$_availableEventLoops['swoole']);
    }

    $loop_name = '';
    foreach (static::$_availableEventLoops as $name=>$class) {
        if (\extension_loaded($name)) {
            $loop_name = $name;
            break;
        }
    }

    if ($loop_name) {
        static::$eventLoopClass = static::$_availableEventLoops[$loop_name];
    } else {
        static::$eventLoopClass =  '\Workerman\Events\Select';
    }
    return static::$eventLoopClass;
}

疑難問題定位

posted @ 2023-03-30 16:26 白開水閱讀(452) 評論(0) 收藏舉報

Fairy Tail