Workerman

它是什么

Workerman 是一个异步事件驱动的 PHP 框架，具备高性能，用于构建快速且可扩展的网络应用。它支持 HTTP、WebSocket、自定义协议、协程以及连接池，能够高效地处理高并发场景。

环境要求

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

安装步骤

composer require workerman/workerman

文档

https://manual.workerman.net

基本使用

构建一个 WebSocket 服务器

<?php

use Workerman\Worker;

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

// Create a Websocket server
$ws_worker = new Worker('websocket://0.0.0.0:2346');

// Emitted when new connection come
$ws_worker->onConnect = function ($connection) {
    echo "New connection\n";
};

// Emitted when data received
$ws_worker->onMessage = function ($connection, $data) {
    // Send hello $data
    $connection->send('Hello ' . $data);
};

// Emitted when connection closed
$ws_worker->onClose = function ($connection) {
    echo "Connection closed\n";
};

// Run 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 processes
$http_worker->count = 4;

// Emitted when data received
$http_worker->onMessage = function ($connection, $request) {
    //$request->get();
    //$request->post();
    //$request->header();
    //$request->cookie();
    //$request->session();
    //$request->uri();
    //$request->path();
    //$request->method();

    // Send data to client
    $connection->send("Hello World");
};

// Run all workers
Worker::runAll();

一个 TCP 服务器

use Workerman\Worker;

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

// #### create socket and listen 1234 port ####
$tcp_worker = new Worker('tcp://0.0.0.0:1234');

// 4 processes
$tcp_worker->count = 4;

// Emitted when new connection come
$tcp_worker->onConnect = function ($connection) {
    echo "New Connection\n";
};

// Emitted when data received
$tcp_worker->onMessage = function ($connection, $data) {
    // Send data to client
    $connection->send("Hello $data \n");
};

// Emitted when connection is closed
$tcp_worker->onClose = function ($connection) {
    echo "Connection closed\n";
};

Worker::runAll();

启用 SSL

<?php

use Workerman\Worker;

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

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

// Create a Websocket server with ssl context.
$ws_worker = new Worker('websocket://0.0.0.0:2346', $context);

// Enable SSL. WebSocket+SSL means that Secure WebSocket (wss://). 
// The similar approaches for Https etc.
$ws_worker->transport = 'ssl';

$ws_worker->onMessage = function ($connection, $data) {
    // Send hello $data
    $connection->send('Hello ' . $data);
};

Worker::runAll();

异步TCP连接（TCP/WS/文本/帧等...）

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

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

$worker = new Worker();
$worker->onWorkerStart = function () {
    // Websocket protocol for client.
    $ws_connection = new AsyncTcpConnection('ws://echo.websocket.org:80');
    $ws_connection->onConnect = function ($connection) {
        $connection->send('Hello');
    };
    $ws_connection->onMessage = function ($connection, $data) {
        echo "Recv: $data\n";
    };
    $ws_connection->onError = function ($connection, $code, $msg) {
        echo "Error: $msg\n";
    };
    $ws_connection->onClose = function ($connection) {
        echo "Connection closed\n";
    };
    $ws_connection->connect();
};

Worker::runAll();

协程

协程用于创建协程，使得可以执行异步任务，以提升并发性能。

<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

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

$worker->eventLoop = Swoole::class; // Or Swow::class or Fiber::class

$worker->onMessage = function (TcpConnection $connection, Request $request) {
    Coroutine::create(function () {
        echo file_get_contents("http://www.example.com/event/notify");
    });
    $connection->send('ok');
};

Worker::runAll();

注意：协程需要 Swoole 扩展或 Swow 扩展，或 Fiber revolt/event-loop，以下同理。

屏障（Barrier）

屏障用于在协程中管理并发和同步。它允许任务并发运行，并等待所有任务完成，确保进程同步。

<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Barrier;
use Workerman\Coroutine;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

// Http Server
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // Or Swow::class or Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
    $barrier = Barrier::create();
    for ($i=1; $i<5; $i++) {
        Coroutine::create(function () use ($barrier, $i) {
            file_get_contents("http://127.0.0.1:8002?task_id=$i");
        });
    }
    // Wait all coroutine done
    Barrier::wait($barrier);
    $connection->send('All Task Done');
};

// Task Server
$task = new Worker('http://0.0.0.0:8002');
$task->onMessage = function (TcpConnection $connection, Request $request) {
    $task_id = $request->get('task_id');
    $message = "Task $task_id Done";
    echo $message . PHP_EOL;
    $connection->close($message);
};

Worker::runAll();

并行处理

并行处理能够同时执行多个任务并收集结果。使用 add 方法添加任务，使用 wait 方法等待任务完成并获取结果。与 Barrier 不同，Parallel 会直接返回每个任务的结果。

<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Parallel;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

// Http Server
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // Or Swow::class or Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
    $parallel = new Parallel();
    for ($i=1; $i<5; $i++) {
        $parallel->add(function () use ($i) {
            return file_get_contents("http://127.0.0.1:8002?task_id=$i");
        });
    }
    $results = $parallel->wait();
    $connection->send(json_encode($results)); // Response: ["Task 1 Done","Task 2 Done","Task 3 Done","Task 4 Done"]
};

// Task Server
$task = new Worker('http://0.0.0.0:8002');
$task->onMessage = function (TcpConnection $connection, Request $request) {
    $task_id = $request->get('task_id');
    $message = "Task $task_id Done";
    $connection->close($message);
};

Worker::runAll();

通道（Channel）

通道是一种协程间通信的机制。一个协程可以向通道中推送数据，而另一个协程可以从通道中取出数据，从而实现在协程之间的同步和数据共享。

<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Channel;
use Workerman\Coroutine;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

// Http Server
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // Or Swow::class or Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
    $channel = new Channel(2);
    Coroutine::create(function () use ($channel) {
        $channel->push('Task 1 Done');
    });
    Coroutine::create(function () use ($channel) {
        $channel->push('Task 2 Done');
    });
    $result = [];
    for ($i = 0; $i < 2; $i++) {
        $result[] = $channel->pop();
    }
    $connection->send(json_encode($result)); // Response: ["Task 1 Done","Task 2 Done"]
};
Worker::runAll();

连接池

连接池用于管理连接或资源池，通过复用资源（例如，数据库连接）来提升性能。它支持获取、归还、创建以及销毁资源。

<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Pool;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

class RedisPool
{
    private Pool $pool;
    public function __construct($host, $port, $max_connections = 10)
    {
        $pool = new Pool($max_connections);
        $pool->setConnectionCreator(function () use ($host, $port) {
            $redis = new \Redis();
            $redis->connect($host, $port);
            return $redis;
        });
        $pool->setConnectionCloser(function ($redis) {
            $redis->close();
        });
        $pool->setHeartbeatChecker(function ($redis) {
            $redis->ping();
        });
        $this->pool = $pool;
    }
    public function get(): \Redis
    {
        return $this->pool->get();
    }
    public function put($redis): void
    {
        $this->pool->put($redis);
    }
}

// Http Server
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // Or Swow::class or Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
    static $pool;
    if (!$pool) {
        $pool = new RedisPool('127.0.0.1', 6379, 10);
    }
    $redis = $pool->get();
    $redis->set('key', 'hello');
    $value = $redis->get('key');
    $pool->put($redis);
    $connection->send($value);
};

Worker::runAll();

自动获取与释放的池化管理机制

<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Context;
use Workerman\Coroutine;
use Workerman\Coroutine\Pool;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

class Db
{
    private static ?Pool $pool = null;
    public static function __callStatic($name, $arguments)
    {
        if (self::$pool === null) {
            self::initializePool();
        }
        // Get the connection from the coroutine context
        // to ensure the same connection is used within the same coroutine
        $pdo = Context::get('pdo');
        if (!$pdo) {
            // If no connection is retrieved, get one from the connection pool
            $pdo = self::$pool->get();
            Context::set('pdo', $pdo);
            // When the coroutine is destroyed, return the connection to the pool
            Coroutine::defer(function () use ($pdo) {
                self::$pool->put($pdo);
            });
        }
        return call_user_func_array([$pdo, $name], $arguments);
    }
    private static function initializePool(): void
    {
        self::$pool = new Pool(10);
        self::$pool->setConnectionCreator(function () {
            return new \PDO('mysql:host=127.0.0.1;dbname=your_database', 'your_username', 'your_password');
        });
        self::$pool->setConnectionCloser(function ($pdo) {
            $pdo = null;
        });
        self::$pool->setHeartbeatChecker(function ($pdo) {
            $pdo->query('SELECT 1');
        });
    }
}

// Http Server
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // Or Swow::class or Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
    $value = Db::query('SELECT NOW() as now')->fetchAll();
    $connection->send(json_encode($value));
};

Worker::runAll();

可用命令

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

性能基准测试

https://www.techempower.com/benchmarks/#section=data-r19&hw=ph&test=plaintext&l=zik073-1r

得到支持

与 Workerman 相关的其他链接

webman
AdapterMan

捐赠

PayPal

许可协议

Workerman 遵循 MIT 协议。