史上最LOW的PHP连接池解决方案

偏执的太偏执、 2022-06-05 08:05 204阅读 0赞

https://huoding.com/2017/09/10/635?utm\_source=tuicool&utm\_medium=referral

大多数 PHP 程序员从来没有使用过连接池，主要原因是按照 PHP 本身的运行机制并不容易实现连接池，于是乎 PHP 程序员一方面不得不承受其它程序员的冷嘲热讽，另一方面还得面对频繁短链接导致的性能低下和 TIME\_WAIT 等问题。 说到这，我猜一定会有 PHP 程序员跳出来说可以使用长连接啊，效果是一样一样的。比如以 PHP 中最流行的 Redis 模块 [PhpRedis][] 为例，便有 [pconnect][] 方法可用，通过它可以复用之前创建的连接，效果和使用连接池差不多。可惜实际情况是 PHP 中各个模块的长连接方法并不好用，基本上是鸡肋一样的存在，原因如下：

*  首先，按照 PHP 的运行机制，长连接在建立之后只能寄居在工作进程之上，也就是说有多少个工作进程，就有多少个长连接，打个比方，我们有 10 台 PHP 服务器，每台启动 1000 个 PHP-FPM 工作进程，它们连接同一个 Redis 实例，那么此 Redis 实例上最多将存在 10000 个长连接，数量完全失控了！
 *  其次，PHP 的长连接本身并不健壮。一旦网络异常导致长连接失效，没有办法自动关闭重新连接，以至于后续请求全部失败，此时除了重启服务别无它法！

问题分析到这里似乎进入了死胡同：按常规做法是没法实现了。 别急着，如果问题比较棘手，我们不妨绕着走。让我们把目光聚焦到 Nginx 的身上，其 [stream][] 模块实现了 TCP/UDP 服务的负载均衡，同时借助 [stream-lua][] 模块，我们就可以实现可编程的 stream 服务，也就是用 Nginx 实现自定义的 TCP/UDP 服务！当然你可以自己从头写 TCP/UDP 服务，不过站在 Nginx 肩膀上无疑是更省时省力的选择。 可是 Nginx 和 PHP 连接池有什么关系？且听我慢慢道来：通常大部分 PHP 是搭配 Nginx 来使用的，而且 PHP 和 Nginx 多半是在同一台服务器上。有了这个客观条件，我们就可以利用 Nginx 来实现一个连接池，在 Nginx 上完成连接 Redis 等服务的工作，然后 PHP 通过本地的 Unix Domain Socket 来连接 Nginx，如此一来既规避了短链接的种种弊端，也享受到了连接池带来的种种好处。

[![PHP Pool][]][PHP Pool]

PHP Pool

下面以 Redis 为例来讲解一下实现过程，事先最好对 Redis 交互协议有一定的了解，推荐阅读[官方文档][Link 1]或[中文翻译][Link 2]，具体实现可以参考 [lua-resty-redis][] 库，虽然它只是一个客户端库，但是 Redis 客户端请求和服务端响应实际上格式是差不多通用的。 首先在 nginx.conf 文件中加入如下配置：

stream {
        lua_code_cache on;
        lua_check_client_abort on;
        lua_package_path "/path/to/?.lua;;";
    
        server {
            listen unix:/tmp/redis.sock;
    
            content_by_lua_block {
                 local redis = require "redis"
                 pool = redis:new({
                     ip = "...", port = "...", auth = "..."
                 })
                 pool:run()
            }
        }
    }

然后在 lua\_package\_path 配置的路径上创建 redis.lua 文件：

local redis = require "resty.redis"
    
    local assert = assert
    local print = print
    local rawget = rawget
    local setmetatable = setmetatable
    local tonumber = tonumber
    local str_byte = string.byte
    local str_gmatch = string.gmatch
    local str_sub = string.sub
    local str_upper = string.upper
    
    local function parse_request(sock)
        local line, err = sock:receive()
    
        if not line then
            return nil, err
        end
    
        local prefix = str_byte(line)
    
        if prefix == 42 then -- char '*'
            local result = {}
    
            local num = tonumber(str_sub(line, 2))
    
            if num <= 0 then
                return nil, "Wrong protocol format"
            end
    
            for i = 1, num do
                local res, err = parse_request(sock)
    
                if res == nil then
                    return nil, err
                end
    
                result[i] = res
            end
    
            return result
        end
    
        if prefix == 36 then -- char '$'
            local size = tonumber(str_sub(line, 2))
    
            if size < 0 then
                return nil, "Wrong protocol format"
            end
    
            local result, err = sock:receive(size)
    
            if not result then
                return nil, err
            end
    
            local crlf, err = sock:receive(2)
    
            if not crlf then
                return nil, err
            end
    
            return result
        end
    
        -- inline
        local result = {}
    
        for res in str_gmatch(line, "%S+") do
            result[#result + 1] = res
        end
    
        return result
    end
    
    local function fetch_response(sock)
        local line, err = sock:receive()
    
        if not line then
            return nil, err
        end
    
        local result = {line, "\r\n"}
        local prefix = str_byte(line)
    
        if prefix == 42 then -- char '*'
            local num = tonumber(str_sub(line, 2))
    
            if num <= 0 then
                return result
            end
    
            for i = 1, num do
                local res, err = fetch_response(sock)
    
                if res == nil then
                    return nil, err
                end
    
                for x = 1, #res do
                    result[#result + 1] = res[x]
                end
            end
        elseif prefix == 36 then -- char '$'
            local size = tonumber(str_sub(line, 2))
    
            if size < 0 then
                return result
            end
    
            local res, err = sock:receive(size)
    
            if not res then
                return nil, err
            end
    
            local crlf, err = sock:receive(2)
    
            if not crlf then
                return nil, err
            end
    
            result[#result + 1] = res
            result[#result + 1] = crlf
        end
    
        return result
    end
    
    local function build_data(value)
        local result = {"*", #value, "\r\n"}
    
        for i = 1, #value do
            local v = value[i]
    
            result[#result + 1] = "$"
            result[#result + 1] = #v
            result[#result + 1] = "\r\n"
            result[#result + 1] = v
            result[#result + 1] = "\r\n"
        end
    
        return result
    end
    
    local function status_reply(message)
        return "+" .. message .. "\r\n"
    end
    
    local function command_args(request)
        local command = request[1]
        command = str_upper(command)
    
        local args = {}
    
        for i = 2, #request do
            args[#args + 1] = request[i]
        end
    
        return command, args
    end
    
    local function exit(err)
        ngx.log(ngx.NOTICE, err)
    
        return ngx.exit(ngx.ERROR)
    end
    
    local _M = {}
    
    _M._VERSION = "1.0"
    
    function _M.new(self, config)
        local t = {
            _ip = config.ip or "127.0.0.1",
            _port = config.port or 6379,
            _timeout = config.timeout or 100000,
            _size = config.size or 10,
            _auth = config.auth,
        }
    
        return setmetatable(t, { __index = _M })
    end
    
    function _M.run(self)
        local ip = self._ip
        local port = self._port
        local timeout = self._timeout
        local size = self._size
        local auth = self._auth
    
        local red = redis:new()
        local ok, err = red:connect(ip, port)
    
        if not ok then
            return exit(err)
        end
    
        if auth then
            local times = assert(red:get_reused_times())
    
            if times == 0 then
                local ok, err = red:auth(auth)
    
                if not ok then
                    return exit(err)
                end
            end
        end
    
        local database = 0
        local transactional = false
    
        local upstream_sock = rawget(red, "_sock")
        local downstream_sock = assert(ngx.req.socket(true))
    
        while true do
            local request, err = parse_request(downstream_sock)
    
            if not request then
                if err == "client aborted" then
                    break
                end
    
                return exit(err)
            end
    
            local command, args = command_args(request)
    
            if command == "QUIT" then
                downstream_sock:send(status_reply("OK"))
                break
            end
    
            upstream_sock:send(build_data(request))
            local response, err = fetch_response(upstream_sock)
    
            if not response then
                return exit(err)
            end
    
            if command == "SELECT" then
                database = tonumber(args[1])
            elseif command == "MULTI" then
                transactional = true
            elseif command == "EXEC" or command == "DISCARD" then
                transactional = false
            end
    
            downstream_sock:send(response)
        end
    
        if database ~= 0 then
            red:select(0)
        end
    
        if transactional then
            red:discard()
        end
    
        red:set_keepalive(timeout, size)
    end
    
    return _M

测试的 PHP 脚本内容如下：

<?php
    
    // 使用连接池
    $redis = new Redis();
    $redis->connect('/tmp/redis.sock');
    $redis->set("foo", bar);
    $redis->get("foo");
    
    ?>
    
    <?php
    
    // 不使用连接池
    $redis = new Redis();
    $redis->connect('ip', 'port');
    $redis->auth('password')
    $redis->set("foo", bar);
    $redis->get("foo");
    
    ?>

推荐在独立服务器上用 ab 测试，需要注意 Nginx 的 worker\_processes 别设置太小，否则并发能力上不来，此外，测试过程中注意观察 tw（TIME\_WAIT） 数量：

shell> ab -k -n 10000 -c 10 http://test/url
    shell> watch -n1 'cat /proc/net/sockstat'

通过引入连接池，connect 本身就变得很快了，而且因为我们在连接池中统一完成了 auth 授权，所以 PHP 代码里不用再执行额外的请求。我在 4 核 8 G 配置的服务器上测试，发现使用连接池后，性能提升了 20% 以上，不过要注意的是，如果 redis 操作比较多，那么使用连接池性能提升可能不明显，这是因为连接池本身需要重复解析请求和响应，抵消了部分好处。当然了，连接池还能实现很多高级功能，比如我们可以在连接池里动态判断当前请求查询的 key 是不是 hot key，是就本地缓存起来，直接用缓存响应请求。

大概说明一下连接池的原理，当我们 connect 的时候，ngx lua 会优先从连接池中获取连接，当我们 set\_keepalive 的时候，ngx lua 会把连接放回连接池。在一次连接里，用户可能需要多次操作 Redis，于是我们使用了 while true 来循环获取用户的多次操作，不过这样的话，需要有一个请求结束的标识，以便跳出循环执行 set\_keepalive，从而把连接放回连接池，最简单的方法无疑是监控客户端关闭连接的事件，对 PHP 来说是很简单，请求结束时自然会关闭连接，如果你希望提前释放连接的话，那么需要一个标识，语义上 QUIT 是很好的标识，用的话可以手动发送一个 rawCommand('quit')。

鲁迅说：真的猛士，敢于直面惨淡的人生，敢于正视淋漓的鲜血。从这个角度看，本文的做法实在是 LOW，不过换个角度看，二战中德军对付马其诺防线也干过类似的勾当，所以管它 LOW 不 LOW，能解决问题的方法就是好方法。

此条目由 [老王][Link 3]发表在 [Technical][]分类目录，并贴了 [PHP][]标签。将 [固定链接][Link 4]加入收藏夹。

[PhpRedis]: https://github.com/phpredis/phpredis
[pconnect]: https://github.com/phpredis/phpredis#pconnect-popen
[stream]: http://nginx.org/en/docs/stream/ngx_stream_core_module.html
[stream-lua]: https://github.com/openresty/stream-lua-nginx-module
[PHP Pool]: /images/20220605/84d0f03911b24884a805009f599efe22.png
[Link 1]: https://redis.io/topics/protocol
[Link 2]: http://redisdoc.com/topic/protocol.html
[lua-resty-redis]: https://github.com/openresty/lua-resty-redis
[Link 3]: https://huoding.com/author/laowang
[Technical]: https://huoding.com/category/technical
[PHP]: https://huoding.com/tag/php
[Link 4]: https://huoding.com/2017/09/10/635