Workerman

这是什么

Workerman 是一个异步事件驱动的 PHP 框架，具有高性能，可用于构建快速且可扩展的网络应用程序。 Workerman 支持 HTTP、WebSocket、SSL 和其他自定义协议。 Workerman 支持事件扩展。

系统要求

兼容 POSIX 的操作系统（Linux、OSX、BSD）需要 POSIX 和 PCNTL 扩展推荐使用 Event 扩展以获得更好的性能

安装

composer require workerman/workerman

文档

https://manual.workerman.net

基本用法

WebSocket 服务器

<?php

use Workerman\Worker;

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

// 创建一个 WebSocket 服务器
$ws_worker = new Worker('websocket://0.0.0.0:2346');

// 当新连接建立时触发
$ws_worker->onConnect = function ($connection) {
    echo "新连接\n";
};

// 当收到数据时触发
$ws_worker->onMessage = function ($connection, $data) {
    // 发送 hello $data
    $connection->send('你好 ' . $data);
};

// 当连接关闭时触发
$ws_worker->onClose = function ($connection) {
    echo "连接关闭\n";
};

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

HTTP 服务器

use Workerman\Worker;

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

// #### http worker ####
$http_worker = new Worker('http://0.0.0.0:2345');

// 4 个进程
$http_worker->count = 4;

// 当收到数据时触发
$http_worker->onMessage = function ($connection, $request) {
    //$request->get();
    //$request->post();
    //$request->header();
    //$request->cookie();
    //$request->session();
    //$request->uri();
    //$request->path();
    //$request->method();

    // 向客户端发送数据
    $connection->send("你好，世界");
};

// 运行所有 worker
Worker::runAll();

TCP 服务器

use Workerman\Worker;

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

// #### 创建 socket 并监听 1234 端口 ####
$tcp_worker = new Worker('tcp://0.0.0.0:1234');

// 4 个进程
$tcp_worker->count = 4;

// 当新连接建立时触发
$tcp_worker->onConnect = function ($connection) {
    echo "新连接\n";
};

// 当收到数据时触发
$tcp_worker->onMessage = function ($connection, $data) {
    // 向客户端发送数据
    $connection->send("你好 $data \n");
};

// 当连接关闭时触发
$tcp_worker->onClose = function ($connection) {
    echo "连接关闭\n";
};

Worker::runAll();

启用 SSL

<?php

use Workerman\Worker;

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

// SSL 上下文
$context = [
    'ssl' => [
        'local_cert'  => '/your/path/of/server.pem',
        'local_pk'    => '/your/path/of/server.key',
        'verify_peer' => false,
    ]
];

// 创建带有 ssl 上下文的 WebSocket 服务器
$ws_worker = new Worker('websocket://0.0.0.0:2346', $context);

// 启用 SSL。WebSocket+SSL 意味着安全 WebSocket（wss://）。
// 类似的方法也适用于 Https 等。
$ws_worker->transport = 'ssl';

$ws_worker->onMessage = function ($connection, $data) {
    // 发送 hello $data
    $connection->send('你好 ' . $data);
};

Worker::runAll();

自定义协议

Protocols/MyTextProtocol.php


namespace Protocols;

/**
 * 用户自定义协议
 * 格式：文本+"\n"
 */
class MyTextProtocol
{
    public static function input($recv_buffer)
    {
        // 查找第一个出现 "\n" 的位置
        $pos = strpos($recv_buffer, "\n");

        // 不是完整的包。返回 0 因为无法计算包的长度
        if ($pos === false) {
            return 0;
        }

        // 返回包的长度
        return $pos + 1;
    }

    public static function decode($recv_buffer)
    {
        return trim($recv_buffer);
    }

    public static function encode($data)
    {
        return $data . "\n";
    }
}

use Workerman\Worker;

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

// #### MyTextProtocol worker ####
$text_worker = new Worker('MyTextProtocol://0.0.0.0:5678');

$text_worker->onConnect = function ($connection) {
    echo "新连接\n";
};

$text_worker->onMessage = function ($connection, $data) {
    // 向客户端发送数据
    $connection->send("你好，世界\n");
};

$text_worker->onClose = function ($connection) {
    echo "连接关闭\n";
};

// 运行所有 worker
Worker::runAll();

定时器


use Workerman\Worker;
use Workerman\Timer;

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

$task = new Worker();
$task->onWorkerStart = function ($task) {
    // 2.5 秒
    $time_interval = 2.5; 
    $timer_id = Timer::add($time_interval, function () {
        echo "定时器运行\n";
    });
};

// 运行所有 worker
Worker::runAll();

异步 TCP 连接（tcp/ws/text/frame 等）


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

require_once __DIR__ . '/vendor/autoload.php';
$worker = new Worker();
$worker->onWorkerStart = function () {
    // 用于客户端的Websocket协议
    $ws_connection = new AsyncTcpConnection('ws://echo.websocket.org:80');
    $ws_connection->onConnect = function ($connection) {
        $connection->send('Hello');
    };
    $ws_connection->onMessage = function ($connection, $data) {
        echo "接收: $data\n";
    };
    $ws_connection->onError = function ($connection, $code, $msg) {
        echo "错误: $msg\n";
    };
    $ws_connection->onClose = function ($connection) {
        echo "连接已关闭\n";
    };
    $ws_connection->connect();
};

Worker::runAll();

使用HTTP代理

<?php

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

use Workerman\Connection\AsyncTcpConnection;
$worker = new \Workerman\Worker();
$worker->onWorkerStart = function($worker){
    echo '开始连接' . PHP_EOL;
    $url = 'ws://stream.binance.com:9443/ws';
    $con = new AsyncTcpConnection($url);
    $con->transport = 'ssl';
//    $con->proxySocks5 = '127.0.0.1:1080';
    $con->proxyHttp = '127.0.0.1:25378';

    $con->onConnect = function(AsyncTcpConnection $con) {
        $ww = [
            'id' => 1,
            'method' => 'SUBSCRIBE',
            'params' => [
                "btcusdt@aggTrade",
                "btcusdt@depth"
            ]
        ];
        echo '连接成功';
        $con->send(json_encode($ww));
        echo '完成';
    };

    $con->onMessage = function(AsyncTcpConnection $con, $data) {
        echo $data;
    };

    $con->onClose = function (AsyncTcpConnection $con) {
        echo '连接关闭' . PHP_EOL;
    };

    $con->onError = function (AsyncTcpConnection $con, $code, $msg) {
        echo "错误 [ $code ] $msg\n";
    };

    $con->connect();
};
\Workerman\Worker::runAll();

使用Socks5代理

<?php

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

use Workerman\Connection\AsyncTcpConnection;
$worker = new \Workerman\Worker();
$worker->onWorkerStart = function($worker){
    echo '开始连接' . PHP_EOL;
    $url = 'ws://stream.binance.com:9443/ws';
    $con = new AsyncTcpConnection($url);
    $con->transport = 'ssl';
    $con->proxySocks5 = '127.0.0.1:1080';
//    $con->proxyHttp = '127.0.0.1:25378';

    $con->onConnect = function(AsyncTcpConnection $con) {
        $ww = [
            'id' => 1,
            'method' => 'SUBSCRIBE',
            'params' => [
                "btcusdt@aggTrade",
                "btcusdt@depth"
            ]
        ];
        echo '连接成功';
        $con->send(json_encode($ww));
        echo '完成';
    };

    $con->onMessage = function(AsyncTcpConnection $con, $data) {
        echo $data;
    };

    $con->onClose = function (AsyncTcpConnection $con) {
        echo '连接关闭' . PHP_EOL;
    };

    $con->onError = function (AsyncTcpConnection $con, $code, $msg) {
        echo "错误 [ $code ] $msg\n";
    };

    $con->connect();
};
\Workerman\Worker::runAll();

代理支持TLS1.3，无需Sniproxy通道

可用命令

php start.php start
php start.php start -d
php start.php status
php start.php status -d
php start.php connections
php start.php stop
php start.php stop -g
php start.php restart
php start.php reload
php start.php reload -g