异步PHP

异步是软件系统不阻塞执行主线程的能力。
异步操作是在完成之前不会阻塞程序执行流的操作。

封锁作业

public function update(User $user)
{
    try {
        $sql = 'UPDATE users SET ...';
        return $this->connection->execute($sql, $user->data());
    } catch (\PDOException $error) {
        log($error->getMessage());
    }

    return 0;
}

• 文件系统：fwrite，file_get_contents。
• 数据库：PDOConnection，RedisClient。默认情况下，几乎所有用于连接数据库的扩展都以阻止模式工作。
• 流程：exec，system，proc_open。这些都是阻塞操作，因为所有处理流程都是通过系统调用构建的。
• 与标准输入/输出工作：readline，echo，print。

异步SQL客户端

public function execAsync(string $query, array $params = [])
{
    $socket = stream_socket_client('127.0.0.1:3306', ...);

    stream_set_blocking($socket, false);

    $data = $this->packBinarySQL($query, $params);
    
    socket_write($socket, $data, strlen($data));
}

诺言

Promise是异步操作结果的包装对象。而且，该操作的结果仍然是我们未知的。

承诺如何运作

interface Promise
{
    const
        STATUS_PENDING = 0,
        STATUS_RESOLVED = 1,
        STATUS_REJECTED = 2
    ;

    public function onResolve(callable $callback);
    public function onReject(callable $callback);
    public function resolve($data);
    public function reject(\Throwable $error);
}

异步SQL客户端

public function execAsync(string $query, array $params = [])
{
    $deferred = new Deferred;

    $socket = stream_socket_client('127.0.0.1:3306', ...);
    stream_set_blocking($socket, false);

    $data = $this->packBinarySQL($query, $params);
    socket_write($socket, $data, strlen($data));

    return $deferred->promise();
}

• 设置回调并获得affected rows返回给我们的回调PDOConnection;
• 处理错误，添加到日志中；
• 如果SQL Server响应错误，请重试查询。

$promise = $this->execAsync($sql, $user->data());

$promise->onResolve(function (int $rows) {
    echo "Affected rows: {$rows}";
});

$promise->onReject(function (\Throwable $error) {
    log($error->getMessage());
});

事件循环

• 客户端告诉事件循环，它对某种套接字感兴趣。
• 事件循环通过系统调用轮询OS stream_select：是否已准备好套接字，是否已写入所有数据，是否是来自另一端的数据。
• 如果操作系统报告套接字未准备好，已阻塞，则事件循环将重复循环。
• 当操作系统通知套接字已准备就绪时，事件循环将控制权返回给客户端并启用（resolve或reject）承诺。

public static function run()
{
    while (true) {
        stream_select($readSockets, $writeSockets, null, 0);
        
        foreach ($readSockets as $i => $socket) {
            call_user_func(self::readCallbacks[$i], $socket);
        }

        // Do same for write sockets
    }
}

public function execAsync(string $query, array $params = [])
{
    $deferred = new Deferred;
    ...
    Loop::onReadable($socket, function ($socket) use ($deferred) {
        $deferred->resolve(socket_read($socket));
    });

    return $deferred->promise();
}

• JavaScript setTimeout setInterval — . N . Event Loop .
• Event Loop . process control, .

Event Loop

ReactPHP

$connection = (new ConnectionFactory)->createLazyConnection();

$promise = $connection->query('UPDATE users SET ...');
$promise->then(
    function (QueryResult $command) {
        echo count($command->resultRows) . ' row(s) in set.';
    },
    function (Exception $error) {
        echo 'Error: ' . $error->getMessage();
    });

    function (QueryResult $command) {
        echo count($command->resultRows) . ' row(s) in set.';
    },

    function (Exception $error) {
        echo 'Error: ' . $error->getMessage();
    });

$promise
    ->then(function ($data) {
        return new Promise(...);
    })
    ->then(function ($data) {
        ...
    }, function ($error) {
        log($error);
    })
    ...

Amphp

$pool = Mysql\pool("host=127.0.0.1 port=3306 db=test");

try {
    $result = yield $pool->query("UPDATE users SET ...");

    echo $result->affectedRows . ' row(s) in set.';
} catch (\Throwable $error) {
    echo 'Error: ' . $error->getMessage();
}

发电机

function generator($counter = 1)
{
    yield $counter++;

    echo "A";

    yield $counter;

    echo "B";

    yield ++$counter;
}

$generator = generator(1);

foreach ($generator as $value) {
    echo $value;
}

while ($generator->valid()) {
    echo $generator->current();

    $generator->next();
}

协程

function printer() {  
    while (true) {     
        echo yield;       
    }                             
}                                

$print = printer();
$print->send('Hello');
$print->send(' PHPRussia');
$print->send(' 2019');
$print->send('!');

function printer() {
    try {
        echo yield;
    } catch (\Throwable $e) {
        echo $e->getMessage();
    }
}

printer()->throw(new \Exception('Ooops...'));

发电机与承诺

class Generator
{
    public function send($data);
    public function throw(\Throwable $error);
}

class Promise
{
    public function resolve($data);
    public function reject(\Throwable $error);
}

function recoil(\Generator $generator)
{
    $promise = $generator->current();

    $promise->onResolve(function($data) use ($generator) {
        $generator->send($data);
        recoil($generator);
    };

    $promise->onReject(function ($error) use ($generator) {
        $generator->throw($error);
        recoil($generator);
    });
}

    $promise->onResolve(function($data) use ($generator) {
        $generator->send($data);
        recoil($generator);
    };

合作多任务

用PHP处理请求

• 简单的错误处理。如果出现问题，并且其中一个请求失败了，我们不需要做任何事情-下一个请求将会到来，这不会影响其工作。
• 我们不考虑记忆。我们不需要清理或监视内存。在下一个请求时，将清除所有内存。

• 限制进程数。如果服务器上有50个FPM线程，则第51个请求到达时，它将等待直到其中一个线程变为空闲。
• 上下文切换的成本。操作系统在FPM流之间切换请求。此处理器级别的操作称为上下文切换。它很昂贵，并采取了大量措施。有必要保存所有寄存器，调用堆栈，处理器中的所有内容，然后切换到另一个进程，加载其寄存器和调用堆栈，在那里再次执行某些操作，再次切换，再次保存...很长时间。

异步HTTP服务器

Loop::run(function () {
    $app = new Application();
    $app->bootstrap();

    $sockets = [Socket\listen('0.0.0.0:80')];

    $server = new Server($sockets, new CallableRequestHandler(
        function (Request $request) use ($app) {
            $response = yield $app->dispatch($request);

            return new Response(Status::OK, [], $response);
        })
    );

    yield $server->start();
});

问题

class UserRepository
{
    public function find(int $id): \Generator
    {
        $data = yield $this->db->query('SELECT ...', $id);

        return User::fill($data);
    }
}

PHP 8

class UserRepository
{
    public function find(int $id): Promise<User>
    {
        $data = await $this->db->query('SELECT ...', $id);

        return User::fill($data);
    }
}

class UserRepository
{
    public function find(int $id): Promise<User>
    {
        $data = yield $this->db->query('SELECT ...', $id);

        return User::fill($data);
    }
}

为什么要这样？

为了速度和性能。

2020年5月13日，我们将第二次聚集在PHP俄罗斯，讨论该语言，库和框架，提高生产率的方式以及炒作解决方案的陷阱。我们已经接受了前4 份报告，但是征文通知仍在进行中。应用，如果你想与社区分享您的体验。