前言

今年接觸了一個(gè)策略類手游相關(guān)的項(xiàng)目，后端本身計(jì)劃是使用skynet進(jìn)行開發(fā)的，后來結(jié)合項(xiàng)目的時(shí)間緊急程度和客戶端開發(fā)組討論后決定使用PHP進(jìn)行快速開發(fā)，后期再使用其他語言框架進(jìn)行拆分業(yè)務(wù)；綜合考慮最后選用了webman作為主要開發(fā)框架。

整體項(xiàng)目分為配置服務(wù)、HTTP-API服務(wù)、websocket服務(wù)三大部分，其中配置管理主要是兼容客戶端生成的配置數(shù)據(jù)進(jìn)行導(dǎo)入導(dǎo)出轉(zhuǎn)換加載，底層使用MySQL進(jìn)行儲(chǔ)存，多服務(wù)間使用Redis進(jìn)行一級(jí)緩存，服務(wù)進(jìn)程間使用了基于APCu的共享緩存，后期我將該共享緩存組件化也貢獻(xiàn)給了社區(qū)。

【workbunny】共享高速緩存 https://www.workerman.net/plugin/133

Redis

在游戲開發(fā)界實(shí)際上使用Redis的情況還是比較多的，我們使用Redis主要還是為了將一些數(shù)據(jù)緩存共享給各個(gè)服務(wù)器實(shí)例:

┌─────┐                                       ┌─────┐
     |  A  | ────────────>  service  <──────────── |  B  |
     └─────┘                                       └─────┘
    /   |   \                                     /   |   \
┌───┐ ┌───┐ ┌───┐                             ┌───┐ ┌───┐ ┌───┐
| a | | b | | c | ───────>  instance <─────── | a | | b | | c |
└───┘ └───┘ └───┘                             └───┘ └───┘ └───┘
  |     |     |                                 |     |     |
 1|2   1|2   1|2 ────────>  process  <──────── 1|2   1|2   1|2
 3|4   3|4   3|4                               3|4   3|4   3|4

如圖所示，我們分為A/B區(qū)服，每個(gè)區(qū)服下可能存在abc不同的服務(wù)器實(shí)例，他們需要共享相同的區(qū)服配置；每個(gè)區(qū)服各自管理自己的數(shù)據(jù)庫數(shù)據(jù)區(qū)域/數(shù)據(jù)庫實(shí)例；每個(gè)區(qū)服下的服務(wù)器實(shí)例對(duì)于數(shù)據(jù)庫數(shù)據(jù)的要求是強(qiáng)需求，且為變動(dòng)較為頻繁的數(shù)據(jù)內(nèi)容，與web的微服務(wù)有區(qū)別，所以我們沒有使用類似Nacos或者其他配置中心進(jìn)行處理，從而用更適配當(dāng)前場(chǎng)景的Redis作為緩存服務(wù)。

同時(shí)Redis也可以作為用戶登錄鑒權(quán)相關(guān)中的一環(huán)，也可以為運(yùn)營相關(guān)功能提供一些輔助，比如使用Redis-Stream作為消息隊(duì)列，處理一些事件通知等。

共享內(nèi)存

在游戲開發(fā)中，許多業(yè)務(wù)都是在內(nèi)存中進(jìn)行的計(jì)算處理，而我們上述的模式是多進(jìn)程模式，進(jìn)程間通訊是一個(gè)比較頻繁出現(xiàn)的點(diǎn)；一開始解決這個(gè)問題是粗暴的將一些固定業(yè)務(wù)固定在對(duì)應(yīng)的進(jìn)程上執(zhí)行，盡可能避免進(jìn)程間的通訊問題。

后來隨著業(yè)務(wù)逐步的擴(kuò)大，單純限制業(yè)務(wù)是沒辦法完全實(shí)現(xiàn)的，這時(shí)候有考慮過使用webman的插件channel；但實(shí)際上channel基于socket涉及系統(tǒng)內(nèi)核態(tài)用戶態(tài)的拷貝等問題，同時(shí)受網(wǎng)絡(luò)影響受限，在一些業(yè)務(wù)的計(jì)算處理上會(huì)帶來比較高的延遲，包括Redis也同樣是這樣的問題，我們需要實(shí)現(xiàn)數(shù)據(jù)的零拷貝。

后續(xù)我們的目標(biāo)鎖定在了共享內(nèi)存上，因?yàn)楣蚕韮?nèi)存可以輕易的在進(jìn)程間進(jìn)行通訊交換，而且不存在深拷貝和網(wǎng)絡(luò)等問題，效率、性能非常的高，整體微秒級(jí)別的響應(yīng)滿足我們的需求；于是我基于PHP的拓展APCu封裝了適合我們業(yè)務(wù)場(chǎng)景的插件包進(jìn)行使用。

webman-shared-cache

我們的基礎(chǔ)應(yīng)用實(shí)現(xiàn)了定時(shí)器來從MySQL數(shù)據(jù)庫讀取配置信息，定時(shí)器的處理器也在讀取數(shù)據(jù)刷入Redis的同時(shí)觸發(fā)共享內(nèi)存的更新事件，上層業(yè)務(wù)通過更新事件的回調(diào)出發(fā)會(huì)將Redis的數(shù)據(jù)刷入共享內(nèi)存中，以便當(dāng)前區(qū)服實(shí)例的各個(gè)進(jìn)程能夠使用。

我們使用緩存的場(chǎng)景很多都是MAP數(shù)據(jù)，所以我在實(shí)現(xiàn)插件的時(shí)候特別實(shí)現(xiàn)了類似Redis-Hash相關(guān)的功能：HSet/HGet/HDel/HKeys/HExists。

由于我們需要一些自增自減的運(yùn)算，所以也實(shí)現(xiàn)了以下功能點(diǎn)：HIncr/HDecr 支持浮點(diǎn)運(yùn)算。由于APCu的特性所以儲(chǔ)存的數(shù)據(jù)也是支持儲(chǔ)存對(duì)象數(shù)據(jù)的；

webman-shared-cache為何使用鎖？

APCu（Alternative PHP Cache User Cache）是一個(gè)開放源代碼的PHP緩存擴(kuò)展，它提供了一種在PHP應(yīng)用程序中存儲(chǔ)和檢索數(shù)據(jù)的快速方法。它是APC（Alternative PHP Cache）的繼任者，專注于用戶數(shù)據(jù)的緩存，而不是opcode緩存。

之前我有和社區(qū)的同學(xué)們聊過，他們不是很理解為什么我在實(shí)現(xiàn)插件的時(shí)候自己使用了鎖，這是因?yàn)锳PCu本身的自行實(shí)現(xiàn)了對(duì)它自身函數(shù)的原子性操作，但我們使用它的時(shí)候是在多進(jìn)程的環(huán)境下，每一個(gè)進(jìn)程內(nèi)存在多次APCu的操作，為了業(yè)務(wù)的原子性，我們希望這多次的操作要在一個(gè)原子性內(nèi)完成，所以需要一個(gè)鎖來進(jìn)行隔離，以免在多進(jìn)程的環(huán)境下被其他進(jìn)程的操作污染，整體是類似MySQl的事務(wù)的：

protected static function _HIncr(string $key, string|int $hashKey, int|float $hashValue = 1): bool|int|float
{
    $func = __FUNCTION__;
    $result = false;
    $params = func_get_args();
    self::_Atomic($key, function () use (
        $key, $hashKey, $hashValue, $func, $params, &$result
    ) {
        $hash = self::_Get($key, []);
        if (is_numeric($v = ($hash[$hashKey] ?? 0))) {
            $hash[$hashKey] = $result = $v + $hashValue;
            self::_Set($key, $hash);
        }
        return [
            'timestamp' => microtime(true),
            'method'    => $func,
            'params'    => $params,
            'result'    => null
        ];
    }, true);
    return $result;
}

比如上述代碼，就是一個(gè)Hash key的自增操作，我們需要在讀取Hash后在寫入，讀取和寫入應(yīng)為一體的；

原子性執(zhí)行函數(shù)Atomic的實(shí)現(xiàn)如下：

/**
     * 原子操作
     *  - 無法對(duì)鎖本身進(jìn)行原子性操作
     *  - 只保證handler是否被原子性觸發(fā)，對(duì)其邏輯是否拋出異常不負(fù)責(zé)
     *  - handler盡可能避免超長阻塞
     *  - lockKey會(huì)被自動(dòng)設(shè)置特殊前綴#lock#，可以通過Cache::LockInfo進(jìn)行查詢
     *
     * @param string $lockKey
     * @param Closure $handler
     * @param bool $blocking
     * @return bool
     */
    protected static function _Atomic(string $lockKey, Closure $handler, bool $blocking = false): bool
    {
        $func = __FUNCTION__;
        $result = false;
        if ($blocking) {
            $startTime = time();
            while ($blocking) {
                // 阻塞保險(xiǎn)
                if (time() >= $startTime + self::$fuse) {return false;}
                // 創(chuàng)建鎖
                apcu_entry($lock = self::GetLockKey($lockKey), function () use (
                    $lockKey, $handler, $func, &$result, &$blocking
                ) {
                    $res = call_user_func($handler);
                    $result = true;
                    $blocking = false;
                    return [
                        'timestamp' => microtime(true),
                        'method'    => $func,
                        'params'    => [$lockKey, '\Closure'],
                        'result'    => $res
                    ];
                });
            }
        } else {
            // 創(chuàng)建鎖
            apcu_entry($lock = self::GetLockKey($lockKey), function () use (
                $lockKey, $handler, $func, &$result
            ) {
                $res = call_user_func($handler);
                $result = true;
                return [
                    'timestamp' => microtime(true),
                    'method'    => $func,
                    'params'    => [$lockKey, '\Closure'],
                    'result'    => $res
                ];
            });
        }
        if ($result) {
            apcu_delete($lock);
        }
        return $result;
    }

當(dāng)使用阻塞模式的時(shí)候，我們會(huì)在當(dāng)前進(jìn)程內(nèi)使用一個(gè)while循環(huán)來進(jìn)行阻塞搶占，為了不將當(dāng)前進(jìn)程阻塞死，我們還加入了一個(gè)保險(xiǎn)，由self::$fuse提供；

注意

這里在實(shí)踐過程中需要注意的是，Atomic在傳入回調(diào)函數(shù)時(shí)切勿再使用匿名函數(shù)作為參數(shù)值或者是通過use傳入一個(gè)匿名函數(shù)，如:

$fuc = function() {
    // do something
}
Cache::Atomic('test', function () use ($fuc) {
    // do anything
})

APCu底層會(huì)對(duì)函數(shù)參數(shù)值或引用參數(shù)進(jìn)行序列化儲(chǔ)存，但匿名函數(shù)不可以被序列化，所以會(huì)拋出一個(gè)異常；但你可以通過當(dāng)前對(duì)象的屬性值或者靜態(tài)屬性來保存一個(gè)匿名函數(shù)，然后在Atomic的回調(diào)內(nèi)調(diào)用使用。

0.4.x版本

由于目前我使用Webman基于SQLite和共享內(nèi)存在自行實(shí)現(xiàn)一個(gè)具備RAFT的輕調(diào)度服務(wù)插件和服務(wù)注冊(cè)與發(fā)現(xiàn)插件，所以特此為其完善增加了Channel特性；

Channel可以輔助實(shí)現(xiàn)類似Redis-List、Redis-stream、Redis-Pub/Sub的功能。

Channel

Channel是個(gè)特殊的數(shù)據(jù)格式，他的格式是固定如下的：

[
    '--default--' => [
        'futureId' => null,
        'value'    => []
    ],
    workerId_1 => [
        'futureId' => 1,
        'value'    => []
    ],
    workerId_2 => [
        'futureId' => 1,
        'value'    => []
    ]，
    ......
]

它在共享內(nèi)存中的鍵默認(rèn)以**#Channel#**開頭。

• --default--是默認(rèn)儲(chǔ)存空間，workerId_1/workerId_2 等是子通道儲(chǔ)存空間，命名是由用戶代碼傳入的，這里建議使用workerman自帶的workerId即可。
• 默認(rèn)儲(chǔ)存空間和子通道儲(chǔ)存空間是互斥的，也就是說當(dāng)存在子通道儲(chǔ)存空間時(shí)，是不存在--default--的，反之亦然；子通道儲(chǔ)存空間是當(dāng)當(dāng)前通道存在監(jiān)聽器時(shí)生成的，而在監(jiān)聽器產(chǎn)生前，消息會(huì)暫存在--default--空間，當(dāng)監(jiān)聽器創(chuàng)建時(shí)，--default--的數(shù)據(jù)value會(huì)被同步到子通道儲(chǔ)存空間內(nèi)，加入value的隊(duì)頭。
• 每一個(gè)子通道儲(chǔ)存空間的value都是拷貝的，存在相同的數(shù)據(jù)，各自監(jiān)聽器監(jiān)聽各自的子通道儲(chǔ)存空間；消息的發(fā)布支持向所有子通道發(fā)布，也可以指定子通道進(jìn)行發(fā)布。
• 監(jiān)聽器的底層使用了workerman的定時(shí)器，區(qū)別與workerman的timer，在event驅(qū)動(dòng)下定時(shí)器的間隔是0，也就是一個(gè)future，而其他的事件驅(qū)動(dòng)是0.001s為間隔。

實(shí)現(xiàn)一個(gè)List

由于監(jiān)聽器創(chuàng)建消費(fèi)是基于workerId的，我們可以通過不同進(jìn)程創(chuàng)建相同的workerId的監(jiān)聽器來對(duì)同一個(gè)子通道進(jìn)行監(jiān)聽：

1. A進(jìn)程使用list作為workerId：

Cache::ChCreateListener('test', 'list', function(string $channelKey, string|int $workerId, mixed $message) {
// TODO 你的業(yè)務(wù)邏輯
});

2. B進(jìn)程也同樣創(chuàng)建list的workerId監(jiān)聽器：

Cache::ChCreateListener('test', 'list', function(string $channelKey, string|int $workerId, mixed $message) {
// TODO 你的業(yè)務(wù)邏輯
});

3. 此時(shí)Channel test的數(shù)據(jù)如下：

[
'list' => [
    'futureId' => 1,
    'value'    => []
],
......
]

注意：共享內(nèi)存中儲(chǔ)存的futureId為最后一個(gè)監(jiān)聽器創(chuàng)建的futureId；當(dāng)當(dāng)前進(jìn)程需要對(duì)監(jiān)聽器進(jìn)行移除時(shí)，請(qǐng)勿使用該數(shù)據(jù)，對(duì)應(yīng)進(jìn)程內(nèi)可以通過Cache::ChCreateListener()的返回值獲取到當(dāng)前進(jìn)程創(chuàng)建的futureId用于移除監(jiān)聽器，不使用共享內(nèi)存中儲(chǔ)存的futureId即可

4. 這時(shí)任意進(jìn)程通過Cache::ChPublish('test', '這是一個(gè)測(cè)試消息'， true);發(fā)送消息，或者指定workerIdCache::ChPublish('test', '這是一個(gè)測(cè)試消息'， true, 'list');。

實(shí)現(xiàn)一個(gè)Pub/Sub

1. A進(jìn)程使用workerman的workerId作為workerId：

Cache::ChCreateListener('test', $worker->id, function(string $channelKey, string|int $workerId, mixed $message) {
// TODO 你的業(yè)務(wù)邏輯
});

2. B進(jìn)程使用workerman的workerId作為workerId：

Cache::ChCreateListener('test', $worker->id, function(string $channelKey, string|int $workerId, mixed $message) {
// TODO 你的業(yè)務(wù)邏輯
});

3. 此時(shí)Channel test的數(shù)據(jù)可能如下：

[
1 => [
    'futureId' => 1,
    'value'    => []
],
2 => [
    'futureId' => 1,
    'value'    => []
]
]

4. 這時(shí)，任意進(jìn)程通過Cache::ChPublish('test', '這是一個(gè)測(cè)試消息'， false);發(fā)送消息即可。

注：發(fā)送消息第三個(gè)參數(shù)使用false時(shí)，如發(fā)送時(shí)還未創(chuàng)建監(jiān)聽器，消息則不會(huì)儲(chǔ)存至Channel，即監(jiān)聽后才可存在消息

實(shí)現(xiàn)類似Redis-stream

與Pub/Sub相同，只不過發(fā)布消息使用Cache::ChPublish('test', '這是一個(gè)測(cè)試消息'， true);， 當(dāng)發(fā)布消息指定workerId時(shí)，可以實(shí)現(xiàn)類似Redis-Stream Group的功能。

注：這里更復(fù)雜的功能可能需要對(duì)workerId進(jìn)行變通，不能簡(jiǎn)單使用workerman自帶的workerId，只需要自行規(guī)劃好即可