PHP异步扩展Swoole笔记(1)

安装Swoole扩展

通过pecl安装, 系统中最好已经有http2依赖, 如果是Ubuntu, 可以直接通过apt安装nghttp2, 如果是Centos或者需要自己编译, 在Github下载nghttp2 https://github.com/tatsuhiro-t/nghttp2 后编译安装) 运行pecl需要autoconf, 如果没有会报错 Cannot find autoconf. Please check your autoconf installation

在Ubuntu下如果php是通过apt安装的, 还需要apt install php-pear 和 php7.2-dev, 否则会报找不到pecl和phpize

sudo pecl install swoole
# 根据自己系统带了哪些模块选择, 我的系统里缺少http2和postgresql, 所以这两个没选
enable sockets supports? [no] : yes
enable openssl support? [no] : yes
enable http2 support? [no] :  
enable mysqlnd support? [no] : yes
enable postgresql coroutine client support? [no] :

然后根据提示, 在php.ini里添加相应的扩展, php.ini的位置可以通过以下命令查看

# php -i |grep php.ini
Configuration File (php.ini) Path => /opt/php/php7.2.10/etc
Loaded Configuration File => /opt/php/php7.2.10/etc/php.ini

;;;;;;;;;;;;;;;;;;;;;;
; Dynamic Extensions ;
;;;;;;;;;;;;;;;;;;;;;;
...
;extension=pdo_sqlite
;extension=pgsql
;extension=shmop
extension=mongodb
extension=swoole

重启php-fpm后, 在phpinfo()里就能看到swoole的信息了. 在命令行下, 可以通过下面的命令查看

php -m | grep swoole

工作模式

Swoole是一个多进程模式的框架(可以类比Nginx的进程模型), 当启动一个Swoole应用时, 一共会创建2 + n + m个进程, 其中n为Worker进程数, m为TaskWorker进程数, 2为一个Master进程和一个Manager进程, 它们之间的关系如下图所示

Master进程为主进程, 该进程会创建Manager进程、Reactor线程等工作进/线程.

Reactor线程实际运行epoll实例，用于accept客户端连接以及接收客户端数据
Manager进程为管理进程，该进程的作用是创建、管理所有的Worker进程和TaskWorker进程

Worker进程作为Swoole的工作进程, 所有的业务逻辑代码均在此进程上运行. 当Reactor线程接收到来自客户端的数据后, 会将数据打包通过管道发送给某个Worker进程. 当一个Worker进程被成功创建后, 会调用onWorkerStart回调, 随后进入事件循环等待数据. 当通过回调函数接收到数据后, 开始处理数据. 如果处理数据过程中出现严重错误导致进程退出, 或者Worker进程处理的总请求数达到指定上限, 则Worker进程调用onWorkerStop回调并结束进程.

基础例子

HTTP Server

简单的http server实现. swoole_http_server不接受onConnect/onReceive/onClose回调设置, 但是额外接受2种新的事件类型onRequest/onMessage

<?php
$http = new swoole_http_server("127.0.0.1", 9501);

$http->on("start", function ($server) {
    echo "Swoole http server is started at http://127.0.0.1:9501
";
});

$http->on("request", function ($request, $response) {
    #print_r($request->header);
    #print_r($request->get);
    #print_r($request->post);
    print_r($request);

    $response->header("Content-Type", "text/plain");
    $response->write(time());
    $response->write(" Hello World
");
    $response->end();
});

$http->start();

关于同步模式和异步模式

在异步模式下, Worker进程内不允许使用任何阻塞式API, 例如MySQL、Redis、http_client、file_get_contents、sleep等。需要使用Swoole提供的各种异步API来实现, 如异步swoole_client, swoole_event_add, swoole_timer, swoole_get_mysqli_sock等API. 官网文档的Http/Server部分没有说明白如何切换这两个模式, 在后面的"高级"部分对这个有解释:

同步阻塞函数

mysql、mysqli、pdo以及其他DB操作函数
sleep、usleep
curl
stream、socket扩展的函数
swoole_client同步模式
memcache、redis扩展函数
file_get_contents/fread等文件读取函数
swoole_server->taskwait
swoole_server->sendwait

swoole_server的PHP代码中有上述函数, Server就是同步服务器, 代码中没有上述函数就是异步服务器

异步非阻塞函数

swoole_client异步模式
mysql-async库
redis-async库
swoole_timer_tick/swoole_timer_after
swoole_event系列函数
swoole_table/swoole_atomic/swoole_buffer
swoole_server->task/finish函数

定时器例子

新版本里面使用的是tick和after方法添加定时任务

<?php

class TimerServer
{
	private $serv;

	public function __construct() {
		$this->serv = new swoole_server("0.0.0.0", 9501);
        	$this->serv->set(array(
	            'worker_num' => 3,
	            'daemonize' => false,
	            'max_request' => 10000,
	            'dispatch_mode' => 2,
	            'debug_mode'=> 1 ,
	        ));
        	$this->serv->on('WorkerStart', array($this, 'onWorkerStart'));
	        $this->serv->on('Connect', array($this, 'onConnect'));
	        $this->serv->on('Receive', array($this, 'onReceive'));
	        $this->serv->on('Close', array($this, 'onClose'));
	        $this->serv->start();
	}

	public function onWorkerStart( $serv , $worker_id) {
        	echo "onWorkerStart
";
	        // 只有当worker_id为0时才添加定时器,避免重复添加
        	if( $worker_id == 0 ) {
			$serv->tick(1000, array($this,'onTimer'), '1');
		}
	}

	public function onConnect( $serv, $fd, $from_id ) {
		echo "Client {$fd} connect
";
	}

	public function onReceive( swoole_server $serv, $fd, $from_id, $data ) {
		echo "Get Message From Client {$fd}:{$data}
";
	}

	public function onClose( $serv, $fd, $from_id ) {
		echo "Client {$fd} close connection
";
	}

	public function onTimer($timer_id, $param) {
		echo 'Timer:'. $timer_id . ' ' . $param . "
";
    	}
}

new TimerServer();

操作Unix Socket的例子

启动和worker数量一样多的socket, 在每次request请求时, 每个worker都使用自己的socket处理请求, 避免出现socket has already been bound to another coroutine错误.
绑定socket前检查socket是否未清除. 如果要停止服务, 使用 kill -15 [master_id] 命令. 如果使用kill -9的话, 不会调用 onWorkerStop 和 onShutdown

<?php

class ShadowManagerServer {
    private $serv;
    private $sockets;

    public function __construct() {
        $this->sockets = array();

        $this->serv = new swoole_http_server('0.0.0.0', 50099);
        $this->serv->set(array(
            'worker_num' => 2,
            'daemonize' => true,
            'max_request' => 1000,
            'dispatch_mode' => 1,
            'reload_async' => true,
            'debug_mode' => 1,
            'log_file' => '/home/milton/shadow_manager.log',
            'log_level' => SWOOLE_LOG_TRACE,
            'trace_flags' => SWOOLE_TRACE_ALL,
        ));

        $this->serv->on('Start', function(swoole_http_server $server) {
            echo '[onStart] PID:'. $server->master_pid . ', workers:' . $server->setting['worker_num'] . "
";
        });

        $this->serv->on('Shutdown', function(swoole_http_server $server) {
            echo '[onShutdown] PID:'. $server->master_pid . "
";
        });

        $this->serv->on('ManagerStart', function(swoole_http_server $server) {
            echo '[onManagerStart] PID:'. $server->master_pid . "
";
        });

        $this->serv->on('ManagerStop', function(swoole_http_server $server) {
            echo '[onManagerStop] PID:'. $server->master_pid . "
";
        });

        $this->serv->on('WorkerStart', function(swoole_http_server $server, int $worker_id) {
            echo '[onWorkerStart] PID:'. $server->master_pid . ', workerId:' . $worker_id . "
";
            $socket = new SwooleCoroutineSocket(AF_UNIX, SOCK_DGRAM, 0);
            if (!$socket) {
                die("socket_create failed
");
                throw new RuntimeException('socket_create failed');
            }
            $tmp_socket_file = '/tmp/ss-php.sock.' . $worker_id;
            # In case the socket file exists
            if (file_exists($tmp_socket_file)) {
                unlink($tmp_socket_file);
            }
            if (!$socket->bind($tmp_socket_file)) {
                throw new RuntimeException('unable to bind to ' . $tmp_socket_file);
            }
            if (!$socket->connect('/var/run/shadowsocks-libev.sock')) {
                throw new RuntimeException("unable to connect to shadowsocks socket");
            }
            $this->sockets[$worker_id] = $socket;
        });

        $this->serv->on('WorkerStop', function(swoole_http_server $server, int $worker_id) {
            echo '[onWorkerStop] PID:'. $server->master_pid . ', workerId:' . $worker_id . "
";
            if (isset($this->sockets[$worker_id])) {
                $socket = $this->sockets[$worker_id];
                $socket->close();
                $tmp_socket_file = '/tmp/ss-php.sock.' . $worker_id;
                if (file_exists($tmp_socket_file)) {
                    unlink($tmp_socket_file);
                }
            }
        });

        $this->serv->on('WorkerExit', function(swoole_http_server $server, int $worker_id) {
            echo '[onWorkerExit] PID:'. $server->master_pid . ', workerId:' . $worker_id . "
";
        });

        $this->serv->on('Connect', function(swoole_http_server $server, int $fd, int $reactor_id) {
            echo '[onConnect] PID:'. $server->master_pid . ', fd:' . $fd . ', reactorId:' . $reactor_id . "
";
        });

        $this->serv->on('Receive', function(swoole_http_server $server, int $fd, int $reactor_id, string $data) {
            echo '[onReceive] PID:'. $server->master_pid . ', fd:' . $fd . ', reactorId:' . $reactor_id . "
";
        });

        $this->serv->on('Close', function(swoole_http_server $server, int $fd, int $reactor_id) {
            echo '[onClose] PID:'. $server->master_pid . ', fd:' . $fd . ', reactorId:' . $reactor_id . "
";
        });

        $this->serv->on('Finish', function(swoole_http_server $server, int $task_id, string $data) {
            echo '[onFinish] PID:'. $server->master_pid . ', taskId:' . $task_id . "
";
        });

        $this->serv->on('WorkerError', function(swoole_http_server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal) {
            echo '[onWorkerError] PID:'. $server->master_pid . ', workerId:' . $worker_id . "
";
            if (isset($this->sockets[$worker_id])) {
                $socket = $this->sockets[$worker_id];
                $socket->close();
                $tmp_socket_file = '/tmp/ss-php.sock.' . $worker_id;
                if (file_exists($tmp_socket_file)) {
                    unlink($tmp_socket_file);
                }
            }
        });

        $this->serv->on('PipeMessage', function(swoole_http_server $server, int $src_worker_id, $message) {
            echo '[onPipeMessage] workerId:'. $server->worker_id . ', srcWorkerId:' . $src_worker_id . "
";
        });

        $this->serv->on('Request', function(swoole_http_request $request, swoole_http_response $response) {
            $worker_id = $this->serv->worker_id;
            echo '[onRequest] PID:' . $this->serv->master_pid . ', workerId:' . $worker_id . ', uri:' . $request->server['request_uri'] . "
";
            if ($request->server['request_uri'] == '/favicon.ico') {
                $response->header("Content-Type", "text/plain");
                $response->status(404);
                $response->end();
                return;
            }
            $msg = 'ping';
            $socket = $this->sockets[$worker_id];
            $socket->send($msg);
            $echo = $socket->recv(1024);
            $response->end('<html><head></head><body>'. $echo .'</body></html>');
        });

        $this->serv->start();
    }
}

new ShadowManagerServer();

守护进程

设置daemonize => 1时, 程序将转入后台作为守护进程运行. 如果不启用守护进程, 当ssh终端退出后, 程序将被终止运行.

启用守护进程后，标准输入和输出会被重定向到 log_file
如果未设置log_file, 将重定向到 /dev/null, 所有打印屏幕的信息都会被丢弃
启用守护进程后, CWD(当前目录)环境变量的值会发生变更, 相对路径的文件读写会出错, PHP程序中必须使用绝对路径.

SSL支持

证书的生成, 参考 https://github.com/LinkedDestiny/swoole-concise-guide/blob/master/book/chapter02/ssl.md 与nginx的证书不太一样.

未验证