5、Workerman 基本使用 - Worker 回调属性

作者: 温新

图书: 【Workerman 基本使用】

阅读: 142

时间: 2024-04-23 12:32:09

hi,我是温新,一名 PHPer

Workerman 中的 Worker 类是整个框架的核心类,它主要负责创建并管理应用服务进程。在 Workerman 框架中,一个 Worker 实例通常代表一个监听特定端口并处理客户端连接的服务进程。

Worker 类的主要作用和功能包括:

  1. 监听端口

    • 开发者可以通过 new Worker('协议://ip:port') 的方式创建一个 Worker 实例来监听指定的网络协议(如 TCP、UDP 或 WebSocket)以及 IP 和端口号。
  2. 多进程管理

    • Worker 支持设置多个子进程 ($worker->count) 来同时处理客户端连接,从而实现高并发。
    • 可以通过 Worker 类进行进程相关的操作,比如进程间的通信、进程重启等。
  3. 事件驱动编程

    • 提供了丰富的回调函数接口,如 onConnect(客户端连接时触发)、onMessage(接收到数据时触发)、onClose(连接关闭时触发)、onError(发生错误时触发)等,允许开发者自定义各种网络事件的处理逻辑。
  4. 守护进程化

    • 提供了将工作进程变为守护进程的方法,使得服务可以在后台长期运行,并且能够在意外退出时自动重启。
  5. 定时任务

    • 通过 Worker 类可以方便地设置定时任务,例如使用内置的 Timer 组件,能在指定时间间隔内执行某个回调函数。
  6. 异步I/O支持

    • Worker 内部基于 event-loop 机制实现了异步 I/O,能够有效地利用系统资源,提高应用程序的性能和吞吐量。
  7. 扩展性

    • Worker 类为开发人员提供了良好的扩展能力,可以根据需要扩展自定义协议或其他组件。
  8. 状态管理

    • 可以获取当前 Worker 进程的状态信息,包括进程 ID、运行状态等,并进行相应管理。
  9. 子进程间共享数据

    • 使用全局变量或 Workerman 自带的数据共享机制,在子进程中存储和传递数据。

不先来了解一下构造函数,不然我们就没得玩了,因此,还是先来了解一下构造函数吧。

Worker::__construct 用于初始化一个 Worker 容器实例,其第一个参数是各种协议:tcp、udp、unix、http、websocket、text 及自定义协议。

接下来在案例中演示这些回调属性,这些案例在同一个文件中进行演示,写完一个案例然后清空代码写一下案例。

onWorkerStart

作用:子进程启动时,触发该回调。

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Request;
require_once __DIR__ . '/vendor/autoload.php';


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

// 设置子进程数量
$worker->count = 3;

// Worker 进程启动时触发此回调
$worker->onWorkerStart = function (Worker $worker) {
    echo 'worker starting' . PHP_EOL;
};

Worker::runAll();

测试运行

$ php worker-construct.php start

Workerman[worker-construct.php] start in DEBUG mode
------------------------------------------- WORKERMAN -------------------------------------------
Workerman version:4.1.15          PHP version:8.2.0           Event-Loop:\Workerman\Events\Event
-------------------------------------------- WORKERS --------------------------------------------
proto   user            worker          listen                processes    status           
tcp     codeing         none            tcp://0.0.0.0:8888    3             [OK]            
-------------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
worker starting
worker starting
worker starting

worker starting 输出了 3 次。开启了 3 个进程,因此会输出 3 次。由此可以得出:若开启了 count 个进程,每个子进程运行一次,则总共会运行 count 次。

onConnect

作用:客户端与服务端完成 TCP 三次握手后触发。

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Request;
require_once __DIR__ . '/vendor/autoload.php';


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

// 设置子进程数量
$worker->count = 3;

// 当客户端与服务端成功建立连接时触发的回调函数
$worker->onConnect = function (TcpConnection $connection) {
    echo '客户端与服务端握手成功:' . $connection->getRemoteIp() . PHP_EOL;
};

Worker::runAll();

测试运行

$ php worker-construct.php start

Workerman[worker-construct.php] start in DEBUG mode
------------------------------------------- WORKERMAN -------------------------------------------
Workerman version:4.1.15          PHP version:8.2.0           Event-Loop:\Workerman\Events\Event
-------------------------------------------- WORKERS --------------------------------------------
proto   user            worker          listen                processes    status           
tcp     codeing         none            tcp://0.0.0.0:8888    3             [OK]            
-------------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
客户端与服务端握手成功:127.0.0.1
客户端与服务端握手成功:127.0.0.1

客户端连接

# 客户端 1
$ telnet 127.0.0.1 8888
# 客户端 2
$ telnet 127.0.0.1 8888
  • 当客户端与服务端完成握手后触发该回调,每一个连接的客户端都会触发一次这个回调;
  • 握手完成之后,此时,服务端无法识别这个连接上来的客户端是谁;
  • udp 不会触发此回调和 onClose 回调。

onWorkerReload

作用:重新加载服务时触发此回调。

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';


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

// 设置子进程数量
$worker->count = 3;
$worker->name = 'workerman';

// Worker 进程启动时触发此回调
$worker->onWorkerStart = function (Worker $worker) {
    echo 'worker starting' . PHP_EOL;
};

// Worker 重载时触发此回调
$worker->onWorkerReload = function(Worker $worker)
{
    foreach($worker->connections as $connection) {
        $connection->send('worker reloading');
    }
};

Worker::runAll();

我们开启 4 个终端进行此次测试,1 个用于启动服务,2 个用于连接服务,1 个用于重载服务。

1)终端 1 用于启动服务

$ php worker-construct.php start

Workerman[worker-construct.php] start in DEBUG mode
------------------------------------------- WORKERMAN -------------------------------------------
Workerman version:4.1.15          PHP version:8.2.0           Event-Loop:\Workerman\Events\Event
-------------------------------------------- WORKERS --------------------------------------------
proto   user            worker          listen                processes    status           
tcp     codeing         workerman       tcp://0.0.0.0:8888    3             [OK]            
-------------------------------------------------------------------------------------------------
Press Ctrl+C to stop. Start success.
worker starting
worker starting
worker starting

2)终端 2 和 3 用于连接服务

# 终端 2
$ telnet 127.0.0.1 8888
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
Connection closed by foreign host.


# 终端 3
$ telnet 127.0.0.1 8888
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
Connection closed by foreign host.

# 连接后都没有断开服务

4)终端 4 重载服务

# 1、查看服务进程号

$ ps aux | grep '[w]orkerman' | awk '{print $2}'
135435 # 主进程
# 剩下 3 个为子进程
135477
135478
135479

# 重载服务
# 同时注意终端 2 和 3 的变化及 终端 1 的服务
$ php worker-construct.php reload

# 再次查看进程号
$ ps aux | grep '[w]orkerman' | awk '{print $2}'
135435 # 主进程没有发生变化,子进程 ID 全变
135802
135803
135804

重载服务端后,终端 2 和 3 的连接被关闭,此时终端 1 的输出如下:

$ php worker-construct.php start
...
Press Ctrl+C to stop. Start success.
worker starting
worker starting
worker starting

# 进程重载之后的输出信息
Workerman[worker-construct.php] reloading
进程重载
进程重载
worker starting
worker starting
worker starting

从输出有结果,我们可以发现如下信息:

  • 重载服务后,子进程是退出重启,因此进程号发生了变化;主进程没有变化;
  • 重载服务后,先执行了 onWorkerReload 回调,因此先输出 进程重载 而后输出 worker starting
  • 重载服务后,会重新调用 onWorkerStart 回调。

如何使用子进程不重启?

设置 reloadable 属性为 false 时,子进程不执行重启。

<?php
...
$worker->name = 'workerman';
// 设置子进程不重启
$worker->reloadable = false;

...

服务重新启动后,使用一个终端来测试

$ ps aux | grep '[w]orkerman' | awk '{print $2}'
139304
139305
139306
139307

# 重载服务
$ php worker-construct.php reload

Workerman[worker-construct.php] reload 

# 再次查看进程号
$ ps aux | grep '[w]orkerman' | awk '{print $2}'
139304
139305
139306
139307

onMessage

作用:收到客户端发送过来的数据时触发。

<?php
/**
 * worker-construct.php
 */

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

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

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

// 设置子进程数量
$worker->count = 3;
// 设置进程名称
$worker->name = 'workerman';

$worker->onMessage = function (TcpConnection $connection, $data) {
    echo '收到客户端发送的数据:' . $data;
};

Worker::runAll();

onClose

作用:客户端断开连接时触发。

<?php
/**
 * worker-construct.php
 */

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

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

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

// 设置子进程数量
$worker->count = 3;
// 设置进程名称
$worker->name = 'workerman';

$worker->onClose = function (TcpConnection $connection) {
    echo '客户端:' . $connection->getRemoteIp() . ' 断开连接';
};

Worker::runAll();

注意:若客户端是断网或断电等情况而断开连接时,由于无法及时发送 TCP 的 FIN,因此,服务端无法知道客户端已经断开,也就无法触发 onClose 回调了。对于这种情况,可以采用应用层心跳解决。

onError

作用:当客户端连接发生错误时触发。

onError 目前有三种错误类型:

  • 1、调用Connection::send由于客户端连接断开导致的失败(紧接着会触发onClose回调) (code:WORKERMAN_SEND_FAIL msg:client closed)
  • 2、发送缓冲区已满,但还在发送数据
  • 3、使用异步连接失败

发送缓冲区满导致的发送失败

这个案例会持续向客户端发送消息直到发送缓冲区满,然后尝试继续发送消息以触发错误。

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker("tcp://0.0.0.0:8888");

$worker->onConnect = function($connection) {
    echo "New connection\n";
    // 发送大量数据以填满发送缓冲区
    for ($i = 0; $i < 1000; $i++) {
        // 发送1MB数据
        $connection->send('x' . str_repeat('0', 1024 * 1024)); 
    }
};

$worker->onError = function($connection, $code, $msg) {
    echo "错误代码为:" . $code . '; 错误消息:' . $msg . PHP_EOL;
};

Worker::runAll();

测试运行

# 启动服务端
start in DEBUG mode
...
New connection
错误代码为:2; 错误消息:send buffer full and drop package
错误代码为:2; 错误消息:send buffer full and drop package

# 客户端连接
$ telnet 127.0.0.1 8888

AsyncTcpConnection 异步连接失败

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
use Workerman\Connection\AsyncTcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker("tcp://0.0.0.0:8080");

// 当有客户端连接时
$worker->onConnect = function ($connection) {
    // 使用 AsyncTcpConnection 进行异步连接
    $asyncConnection = new AsyncTcpConnection("tcp://127.0.0.1:8888");

    $asyncConnection->onError = function ($conn, $code, $msg) {
        echo '异步连接出错:' . $code . ';错误消息:' . $msg . PHP_EOL;
    };

    $asyncConnection->connect();
};

// 开始运行工作进程
Worker::runAll();

测试

# 服务端
$ php worker-construct.php start
...
Press Ctrl+C to stop. Start success.
异步连接出错:1;错误消息:connect 127.0.0.1:8081 fail after 0.0001 seconds

# 客户端连接
$ telnet 127.0.0.1 8080
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.

客户端断开连接

<?php
/**
 * worker-construct.php
 */

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

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

$worker = new Worker("tcp://0.0.0.0:8888");

$worker->onError = function(TcpConnection $connection, $code, $msg) {
    echo "error $code $msg\n";
};

Worker::runAll();

onBufferFull

作用:缓冲区被填满时触发

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker("tcp://0.0.0.0:8888");



$worker->onConnect = function(TcpConnection $connection) {
    // 设置当前连接发送缓冲区,单位字节
    $connection->maxSendBufferSize = 512;

    // 发送大量数据以填满发送缓冲区
    for ($i = 0; $i < 10; $i++) {
        // 发送 1MB 数据
        $connection->send('x' . str_repeat('0', 1024 * 1024));
    }
};

$worker->onBufferFull = function (TcpConnection $connection) {
    echo '缓冲区已满' . PHP_EOL;
};

Worker::runAll();

注意事项看文档,有详细介绍。

onBufferDrain

作用:该回调在应用层发送缓冲区数据全部发送完毕后触发

<?php
/**
 * worker-construct.php
 */

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';

$worker = new Worker("tcp://0.0.0.0:8888");



$worker->onConnect = function(TcpConnection $connection) {
    $connection->maxSendBufferSize = 1;
    for ($i = 0; $i < 10; $i++) {
        // 发送 1MB 数据
        $connection->send('x' . str_repeat('0', 1024 * 1024));
    }
};

$worker->onBufferFull = function($connection) {
    echo "Buffer full";
};

$worker->onBufferDrain = function(TcpConnection $connection){
    echo "buffer drain and continue send\n";
};

Worker::runAll();
请登录后再评论