首发原文链接：Swoole 源码分析之 WebSocket 模块
大家好，我是码农先森。
引言
WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。它允许客户端和服务器之间进行实时数据传输。
与传统的 HTTP 请求-响应模型不同，WebSocket 可以保持双向通信通道，从而使得服务器能够主动向客户端推送数据。
Swoole 中的 WebSocket 服务
下面这段代码是从Swoole 官方网站上的引用，从代码中可以看出创建了一个 WebScoket 对象且设置对应的 IP 地址及监听端口，同时还设置了四个回调方法处理对应的事件。
最后，调用 $server->start() 真正的启动 WebScoket 服务。
$server = new Swoole\Websocket\Server('127.0.0.1', 9502);$server->on('start', function ($server) {echo "Websocket Server is started at ws://127.0.0.1:9502\n";
});$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();
那么接下来，我们就从源码角度来分析 Swoole 对 WebSocket 的实现。
源码拆解
这个函数的主要作用是启动 Server 服务。
static void php_swoole_server_onStart(Server *serv) {// 锁定 Server 对象操作serv->lock();// 从 Server 对象中获取到 onStart 回调函数zval *zserv = (zval *) serv->private_data_2;ServerObject *server_object = server_fetch_object(Z_OBJ_P(zserv));auto fci_cache = server_object->property->callbacks[SW_SERVER_CB_onStart];...// 通过 zend::function::call 调用 PHP 层注册的 onStart 处理函数，并传递参数if (fci_cache && UNEXPECTED(!zend::function::call(fci_cache, 1, zserv, nullptr, serv->is_enable_coroutine()))) {php_swoole_error(E_WARNING, "%s->onStart handler error", SW_Z_OBJCE_NAME_VAL_P(zserv));}// 解锁 Server 对象操作serv->unlock();
}
这个函数主要作用是 WebSocket 服务针对客户端建立连接时事件的处理。
void swoole_websocket_onOpen(Server *serv, HttpContext *ctx) {// 通过 session_id 获取与特定客户端连接相关的 Connection 对象Connection *conn = serv->get_connection_by_session_id(ctx->fd);if (!conn) {swoole_error_log(SW_LOG_TRACE, SW_ERROR_SESSION_NOT_EXIST, "session[%ld] is closed", ctx->fd);return;}// Server 对象中获取在 PHP 层设置的回调函数 onOpen。zend_fcall_info_cache *fci_cache = php_swoole_server_get_fci_cache(serv, conn->server_fd, SW_SERVER_CB_onOpen);if (fci_cache) {zval args[2];args[0] = *((zval *) serv->private_data_2);args[1] = *ctx->request.zobject;// 通过 zend::function::call 调用 PHP 层注册的 onOpen 处理函数，并传递参数if (UNEXPECTED(!zend::function::call(fci_cache, 2, args, nullptr, serv->is_enable_coroutine()))) {php_swoole_error(E_WARNING, "%s->onOpen handler error", ZSTR_VAL(swoole_websocket_server_ce->name));serv->close(ctx->fd, false);}}
}
这个函数主要作用是 WebSocket 服务器针对客户端发送消息事件的处理。
int swoole_websocket_onMessage(Server *serv, RecvData *req) {SessionId fd = req->info.fd;uchar flags = 0;zend_long opcode = 0;// 从接收到的数据中获取客户端的 session_id，并根据 session_id 获取对应的端口信息auto port = serv->get_port_by_session_id(fd);if (!port) {return SW_ERR;}zval zdata;char frame_header[2];// 从接收到的数据中解析出 WebSocket 消息的帧头信息和消息内容memcpy(frame_header, &req->info.ext_flags, sizeof(frame_header));php_swoole_get_recv_data(serv, &zdata, req);// 解析出 WebSocket 消息的标志位和操作码flags = frame_header[0];opcode = frame_header[1];// 根据操作码和服务的设置，判断是否需要特殊处理 Close、Ping 或 Pong 类型的消息if ((opcode == WebSocket::OPCODE_CLOSE && !port->open_websocket_close_frame) ||(opcode == WebSocket::OPCODE_PING && !port->open_websocket_ping_frame) ||(opcode == WebSocket::OPCODE_PONG && !port->open_websocket_pong_frame)) {if (opcode == WebSocket::OPCODE_PING) {...}zval_ptr_dtor(&zdata);return SW_OK;}...// Server 对象中获取在 PHP 层设置的回调函数 onMessagezend_fcall_info_cache *fci_cache =php_swoole_server_get_fci_cache(serv, req->info.server_fd, SW_SERVER_CB_onMessage);zval args[2];args[0] = *(zval *) serv->private_data_2;// 构造一个 WebSocket 消息帧的数据结构，并将结果存储在 args[1]php_swoole_websocket_construct_frame(&args[1], opcode, &zdata, flags);zend_update_property_long(swoole_websocket_frame_ce, SW_Z8_OBJ_P(&args[1]), ZEND_STRL("fd"), fd);// 通过 zend::function::call 调用 PHP 层注册的 onMessage 处理函数，并传递相应参数if (UNEXPECTED(!zend::function::call(fci_cache, 2, args, nullptr, serv->is_enable_coroutine()))) {php_swoole_error(E_WARNING, "%s->onMessage handler error", ZSTR_VAL(swoole_websocket_server_ce->name));serv->close(fd, false);}// 释放 zdata 和 args[1] 占用的内存zval_ptr_dtor(&zdata);zval_ptr_dtor(&args[1]);return SW_OK;
}
这个函数的主要作用是关闭 Server 服务。
void php_swoole_server_onClose(Server *serv, DataHead *info) {...// Server 对象中获取在 PHP 层设置的回调函数 onCloseauto *fci_cache = php_swoole_server_get_fci_cache(serv, info->server_fd, SW_SERVER_CB_onClose);Connection *conn = serv->get_connection_by_session_id(session_id);if (!conn) {return;}// 检查当前的 WebSocket 连接状态是否为非活动状态if (conn->websocket_status != swoole::websocket::STATUS_ACTIVE) {// 获取与当前连接相关的监听端口信息ListenPort *port = serv->get_port_by_server_fd(info->server_fd);// 如果该端口开启了 WebSocket 协议，且设置了 onDisconnect 回调函数if (port && port->open_websocket_protocol &&php_swoole_server_isset_callback(serv, port, SW_SERVER_CB_onDisconnect)) {// 获取 onDisconnect 回调函数fci_cache = php_swoole_server_get_fci_cache(serv, info->server_fd, SW_SERVER_CB_onDisconnect);}}if (fci_cache) {...// 通过 zend::function::call 调用 PHP 层注册的 onDisconnect 处理函数，并传递相应参数if (UNEXPECTED(!zend::function::call(fci_cache, argc, args, nullptr, serv->enable_coroutine))) {php_swoole_error(E_WARNING, "%s->onClose handler error", SW_Z_OBJCE_NAME_VAL_P(zserv));}...}...}
这个函数的作用是断开 WebSocket 客户端的连接，并发送关闭帧。
static PHP_METHOD(swoole_websocket_server, disconnect) {// 从 ZEND_THIS 中获取 Server 对象Server *serv = php_swoole_server_get_and_check_server(ZEND_THIS);...// 清空全局的 WebSocket 缓冲区swoole_websocket_buffer->clear();// 将关闭帧数据打包到 WebSocket 缓冲区中if (WebSocket::pack_close_frame(swoole_websocket_buffer, code, data, length, 0) < 0) {RETURN_FALSE;}// 调用 swoole_websocket_server_close 函数来关闭客户端连接，并返回结果RETURN_BOOL(swoole_websocket_server_close(serv, fd, swoole_websocket_buffer, 1));
}
这个函数的作用是在 WebSocket 服务中关闭客户端连接的操作。
static sw_inline bool swoole_websocket_server_close(Server *serv, SessionId fd, String *buffer, bool real_close) {// 尝试将数据推送给客户端，用于判断是否已经关闭连接bool ret = swoole_websocket_server_push(serv, fd, buffer);if (!ret || !real_close) {return ret;}// 获取到客户端连接相关的 Connection 对象Connection *conn = serv->get_connection_by_session_id(fd);if (conn) {// 将该连接的 websocket_status 改变为 WebSocket::STATUS_CLOSINGconn->websocket_status = WebSocket::STATUS_CLOSING;// 立即关闭连接return serv->close(fd, false);} else {return false;}
}
总结
- 在 Swoole 中 WebSocket 服务是继承于 Http 服务。
- 在实际的使用过程中是通过 Http 服务来握手升级成 WebSocket 服务。
- WebSocket 协议的出现解决了通过传统轮询方式来通信的效率问题。
- 同时也为 PHP 在双向通信解决方式上提供了新的解决方案。
