Асинхронный WEB в 2018. Пишем чат на Websocket используя Swoole

Тема Websocket`ов ^[1] уже не раз затрагивалась на Хабре, в частности рассматривались варианты реализации на PHP. Однако, с момента выхода последней статьи ^[2] с обзором разных технологий прошло уже более года, а миру PHP есть чем ^[3] похвастаться за прошедшее время.

В данной статье я хочу представить русскоязычному сообществу Swoole ^[4] — Асинхронный Open Source фреймворк для PHP, написанный на Си, и поставляемый в виде pecl-расширения.

Посмотреть получившееся в итоге приложение(чат) можно: здесь ^[5].
Исходники на github ^[6].

Почему Swoole?

Наверняка найдутся люди, которые будут в принципе против использования PHP для таких целей, однако в пользу PHP часто могут играть:

• Нежелание разводить зоопарк различных языков на проекте
• Возможность использования уже наработанной кодовой базы(если проект на PHP).

Тем не менее, даже сравнивая с node.js/go/erlang и другими языками, нативно предлагающими асинхронную модель, Swoole — фреймворк написанный на Си и объеденивший в себе низкий порог вхождения и мощную функциональность может быть вполне хорошим кандидатом.

Возможности фреймворка:

• Событийная, асинхронная модель программирования
• Асинхронные TCP / UDP / HTTP / Websocket / HTTP2 клиентские/серверные API
• Поддержка IPv4 / IPv6 / Unixsocket / TCP/ UDP и SSL / TLS
• Быстрая сериализация / десериализация данных
• Высокая производительность, расширяемость, поддержка до 1 миллиона одновременных соединений
• Планировщик заданий с точностью до миллисекунд
• Open source ^[7]
• Поддержка сопрограмм(Coroutines) ^[8]

Возможные варианты использования:

• Микросервисы
• Игровые сервера
• Интернет вещей
• Живые системы общения
• WEB API
• Любые другие сервисы от которых требуется моментальный ответ/высокая скорость/асинхронное выполнение

Примеры кода можно увидеть на главной странице сайта ^[9]. В разделе документации более подробная информация о всём функционале фреймворка.

Приступим к использованию

Ниже я опишу процесс написания несложного Websocket сервера для онлайн-чата и возможные при этом затруднения.

Перед тем как начать: Более подробная информация о классах swoole_websocket_server ^[10] и swoole_server ^[11] (Второй класс наследуется от первого).
Исходники самого чата. ^[6]

Для использования автокомплита в IDE предлагается использовать ide-helper ^[12]

Минимальный шаблон Websocket-сервера:

<?php
$server = new swoole_websocket_server("127.0.0.1", 9502);

$server->on('open', function($server, $req) {
    echo "connection open: {$req->fd}n";
});

$server->on('message', function($server, $frame) {
    echo "received message: {$frame->data}n";
    $server->push($frame->fd, json_encode(["hello", "world"]));
});

$server->on('close', function($server, $fd) {
    echo "connection close: {$fd}n";
});

$server->start();

$fd — идентификатор подключения.
Получить текущие подключения:

$server->connections;

Внутри $frame содержаться все отправленные данные. Вот пример пришедшего объекта в функцию onMessage:

SwooleWebSocketFrame Object
(   
    [fd] => 20
    [data] => {"type":"login","username":"new user"}
    [opcode] => 1
    [finish] => 1
)

Данные клиенту отправляются с помощью функции

Server::push($fd, $data, $opcode=null, $finish=null)

Подробнее про фреймы и opcodes на русском — на learn.javascript ^[13]. Раздел «формат данных»

Максимально подробно про протокол Websocket — RFC ^[14]

А как сохранять данные пришедшие на сервер?
Swoole представляет функционал для асинхронной работы с MySQL ^[15], Redis ^[16], файловый ввод-вывод ^[17]

А также swoole_buffer ^[18], swoole_channel ^[19] и swoole_table ^[20]
Думаю различия понять не сложно по документации. Для хранения имён пользователей я выбрал swoole_table. Сами сообщения хранятся в MySQL.

Итак, инициализация таблицы имён пользователей:

        
$users_table = new swoole_table(131072);
$users_table->column('id', swoole_table::TYPE_INT, 5);
$users_table->column('username', swoole_table::TYPE_STRING, 64);
$users_table->create();

Заполнение данными происходит так:

$count = count($messages_table);

$dateTime = time();
$row = ['username' => $username, 'message' => $data->message, 'date_time' => $dateTime];
$messages_table->set($count, $row);

Для работы с MySQL я решил пока не использовать асинхронную модель, а обращаться стандартным способом, из вебсокет-сервера, через PDO

/**
     * @return Message[]
     */
    public function getAll()
    {
        $stmt = $this->pdo->query('SELECT * from messages');
        $messages = [];
        foreach ($stmt->fetchAll() as $row) {
            $messages[] = new Message( $row['username'], $row['message'], new DateTime($row['date_time']) );
        }
        return $messages;
    }

Websocket сервер было решено оформить в виде класса, и стартовать его в конструкторе:

public function __construct()
    {
        $this->ws = new swoole_websocket_server('0.0.0.0', 9502);

        $this->ws->on('open', function ($ws, $request) {
            $this->onConnection($request);
        });
        $this->ws->on('message', function ($ws, $frame) {
            $this->onMessage($frame);
        });
        $this->ws->on('close', function ($ws, $id) {
            $this->onClose($id);
        });

        $this->ws->on('workerStart', function (swoole_websocket_server $ws) {
            $this->onWorkerStart($ws);
        });

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

Возникшие проблемы:

1. У пользователя подключенного к чату обрывается соединение через 60 секунд если не происходит обмена пакетами(т.е. пользователь ничего не отправлял и ничего не получал)
2. Вебсервер теряет соединение с MySQL если долго не происходит никакого взаимодействия

Решение:

В обоих случая нужна реализация функции «пинг», которая будет постоянно каждые n секунд пинговать клиента в первом случае, и базу MySQL во втором.

Так как обе функции должны работать асинхронно, их нужно вызвать в дочерних процессах сервера.

Для этого их можно инициализировать при событии «workerStart». Мы уже определили его в конструкторе, и при этом событии уже вызывается метод $this->onWorkerStart:
Протокол Websocket поддерживает ping-pong ^[21] из коробки. Ниже можно увидеть реализацию на Swoole.

private function onWorkerStart(swoole_websocket_server $ws)
    {
        $this->messagesRepository = new MessagesRepository();

        $ws->tick(self::PING_DELAY_MS, function () use ($ws) {
            foreach ($ws->connections as $id) {
                $ws->push($id, 'ping', WEBSOCKET_OPCODE_PING);
            }
        });
    }

Далее я реализовал простенькую функцию для пинга MySQL сервера каждые N секунд, используя swooleTimer:

    /**
     * Init new Connection, and ping DB timer function
     */
    private static function initPdo()
    {
        if (self::$timerId === null || (!Timer::exists(self::$timerId))) {
            self::$timerId = Timer::tick(self::MySQL_PING_INTERVAL, function () {
                self::ping();
            });
        }

        self::$pdo = new PDO(self::DSN, DBConfig::USER, DBConfig::PASSWORD, self::OPT);
    }

    /**
     * Ping database to maintain the connection
     */
    private static function ping()
    {
        try {
            self::$pdo->query('SELECT 1');
        } catch (PDOException $e) {
            self::initPdo();
        }
    }

Основная часть работы заключалась в написании логики для добавления, сохранения, отправки сообщений(не сложнее обычного CRUD), а далее огромный простор для усовершенствований.

Пока что я привёл свой код к более-менее читаемому виду и объектно-ориентированному стилю, реализовал немного функционала:

— Вход по имени;

/**
     * @param string $username
     * @return bool
     */
    private function isUsernameCurrentlyTaken(string $username) {
        foreach ($this->usersRepository->getByIds($this->ws->connection_list()) as $user) {
            if ($user->getUsername() == $username) {
                return true;
            }
        }
        return false;
    }

<?php

namespace AppHelpers;

use SwooleChannel;

class RequestLimiter
{
    /**
     * @var Channel
     */
    private $userIds;

    const MAX_RECORDS_COUNT = 10;

    const MAX_REQUESTS_BY_USER = 4;

    public function __construct() {
        $this->userIds = new Channel(1024 * 64);
    }

    /**
     * Check if there are too many requests from user
     *  and make a record of request from that user
     *
     * @param int $userId
     * @return bool
     */
    public function checkIsRequestAllowed(int $userId) {
        $requestsCount = $this->getRequestsCountByUser($userId);
        $this->addRecord($userId);
        if ($requestsCount >= self::MAX_REQUESTS_BY_USER) return false;
        return true;
    }

    /**
     * @param int $userId
     * @return int
     */
    private function getRequestsCountByUser(int $userId) {
        $channelRecordsCount = $this->userIds->stats()['queue_num'];
        $requestsCount = 0;

        for ($i = 0; $i < $channelRecordsCount; $i++) {
            $userIdFromChannel = $this->userIds->pop();
            $this->userIds->push($userIdFromChannel);
            if ($userIdFromChannel === $userId) {
                $requestsCount++;
            }
        }

        return $requestsCount;
    }

    /**
     * @param int $userId
     */
    private function addRecord(int $userId) {
        $recordsCount = $this->userIds->stats()['queue_num'];

        if ($recordsCount >= self::MAX_RECORDS_COUNT) {
            $this->userIds->pop();
        }

        $this->userIds->push($userId);
    }
}

— Простенькую защиту от XSS используя ezyang/htmlpurifier ^[22]

<?php

namespace AppHelpers;

class SpamFilter
{
    /**
     * @var string[] errors
     */
    private $errors = [];

    /**
     * @param string $text
     * @return bool
     */
    public function checkIsMessageTextCorrect(string $text) {
        $isCorrect = true;
        if (empty(trim($text))) {
            $this->errors[] = 'Empty message text';
            $isCorrect = false;
        }
        return $isCorrect;
    }

    /**
     * @return string[] errors
     */
    public function getErrors(): array {
        return $this->errors;
    }
}

Frontend у чата пока что весьма сырой, т.к. меня больше привлекает backend, но когда будет больше времени я постараюсь сделать его поприятнее.

Где брать информацию, узнавать новости о фреймворке?

• Английский официальный сайт ^[4] — полезные ссылки, актуальная документация, немного комментариев от пользователей
• Twitter ^[23] — актуальные новости, полезные ссылки, интересные статьи
• Issue tracker(Github) ^[24] — баги, вопросы, общение с создателями фреймворка. Отвечают очень шустро(на мою issue с вопросом ответили за пару часов, помогли с реализацией pingloop).
• Закрытые issues ^[25] — так же советую. Большая база вопросов от пользователей и ответы от создателей фремворка.
• Тесты, написанные разработчиками ^[26] — практически на каждый модуль из документации есть тесты написанные на PHP, показывающие варианты использования.
• Китайская wiki фреймворка ^[27] — вся информация что и в английской, но значительно больше комментариев от пользователей (гугл переводчик в помощь).

API documentation ^[28] — описание некоторых классов и функций фреймворка в довольно удобном виде.

Резюме

Мне кажется, что Swoole очень активно развивался последний год, вышел из стадии когда его можно было назвать «сырым», и теперь вполне составляет конкуренцию использованию node.js/go с точки зрения асинхронного программирования и реализации сетевых протоколов.

Буду рад услышать различные мнения по теме и отзывы от тех кто уже имеет опыт использования Swoole

Пообщаться в описанном чатике можно по ссылке ^[5]
Исходники доступны на Github ^[6].

Автор: EvgeniiR

Источник ^[29]