前言

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

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

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

Redis

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

┌─────┐                                       ┌─────┐
     |  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ū)服，每個區(qū)服下可能存在abc不同的服務器實例，他們需要共享相同的區(qū)服配置；每個區(qū)服各自管理自己的數(shù)據(jù)庫數(shù)據(jù)區(qū)域/數(shù)據(jù)庫實例；每個區(qū)服下的服務器實例對于數(shù)據(jù)庫數(shù)據(jù)的要求是強需求，且為變動較為頻繁的數(shù)據(jù)內容，與web的微服務有區(qū)別，所以我們沒有使用類似Nacos或者其他配置中心進行處理，從而用更適配當前場景的Redis作為緩存服務。

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

共享內存

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

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

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

webman-shared-cache

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

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

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

webman-shared-cache為何使用鎖？

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

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

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;
}

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

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

/**
     * 原子操作
     *  - 無法對鎖本身進行原子性操作
     *  - 只保證handler是否被原子性觸發(fā)，對其邏輯是否拋出異常不負責
     *  - handler盡可能避免超長阻塞
     *  - lockKey會被自動設置特殊前綴#lock#，可以通過Cache::LockInfo進行查詢
     *
     * @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) {
                // 阻塞保險
                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;
    }

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

注意

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

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

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

0.4.x版本

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

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

Channel

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

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

它在共享內存中的鍵默認以**#Channel#**開頭。

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

實現(xiàn)一個List

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

1. A進程使用list作為workerId：

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

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

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

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

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

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

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

實現(xiàn)一個Pub/Sub

1. A進程使用workerman的workerId作為workerId：

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

2. B進程使用workerman的workerId作為workerId：

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

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

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

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

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

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

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

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