phpapcu机能技巧_APCu高速共享缓存插件分享机能超越Redis达10倍

整体项目分为配置做事、HTTP-API做事、websocket做事三大部分，个中配置管理紧张是兼容客户端天生的配置数据进行导入导出转换加载，底层利用MySQL进行储存，多做事间利用Redis进行一级缓存，做事进程间利用了基于APCu的共享缓存，后期我将该共享缓存组件化也贡献给了社区。

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

在游戏开拓界实际上利用Redis的情形还是比较多的，我们利用Redis紧张还是为了将一些数据缓存共享给各个做事器实例:

┌─────┐ ┌─────┐ | 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区服，每个区服下可能存在abc不同的做事器实例，他们须要共享相同的区服配置；每个区服各自管理自己的数据库数据区域/数据库实例；每个区服下的做事器实例对付数据库数据的哀求是强需求，且为变动较为频繁的数据内容，与web的微做事有差异，以是我们没有利用类似Nacos或者其他配置中央进行处理，从而用更适配当前场景的Redis作为缓存做事。

同时Redis也可以作为用户登录鉴权干系中的一环，也可以为运营干系功能供应一些赞助，比如利用Redis-Stream作为行列步队，处理一些事宜关照等。

在游戏开拓中，许多业务都是在内存中进行的打算处理，而我们上述的模式是多进程模式，进程间通讯是一个比较频繁涌现的点；一开始办理这个问题是粗暴的将一些固定业务固定在对应的进程上实行，尽可能避免进程间的通讯问题。

后来随着业务逐步的扩大，纯挚限定业务是没办法完备实现的，这时候有考虑过利用webman的插件channel；但实际上channel基于socket涉及系统内核态用户态的拷贝等问题，同时受网络影响受限，在一些业务的打算处理上会带来比较高的延迟，包括Redis也同样是这样的问题，我们须要实现数据的零拷贝。

后续我们的目标锁定在了共享内存上，由于共享内存可以轻易的在进程间进行通讯交流，而且不存在深拷贝和网络等问题，效率、性能非常的高，整体微秒级别的相应知足我们的需求；于是我基于PHP的拓展APCu封装了适宜我们业务场景的插件包进行利用。

我们的根本运用实现了定时器来从MySQL数据库读取配置信息，定时器的处理器也在读取数据刷入Redis的同时触发共享内存的更新事宜，上层业务通过更新事宜的回调出发会将Redis的数据刷入共享内存中，以便当前区服实例的各个进程能够利用。

我们利用缓存的场景很多都是MAP数据，以是我在实现插件的时候特殊实现了类似Redis-Hash干系的功能：HSet/HGet/HDel/HKeys/HExists。

由于我们须要一些自增自减的运算，以是也实现了以下功能点：HIncr/HDecr 支持浮点运算。
由于APCu的特性以是储存的数据也是支持储存工具数据的；

APCu（Alternative PHP Cache User Cache）是一个开放源代码的PHP缓存扩展，它供应了一种在PHP运用程序中存储和检索数据的快速方法。
它是APC（Alternative PHP Cache）的继任者，专注于用户数据的缓存，而不是opcode缓存。

之前我有和社区的同学们聊过，他们不是很理解为什么我在实现插件的时候自己利用了锁，这是由于APCu本身的自行实现了对它自身函数的原子性操作，但我们利用它的时候是在多进程的环境下，每一个进程内存在多次APCu的操作，为了业务的原子性，我们希望这多次的操作要在一个原子性内完成，以是须要一个锁来进行隔离，以免在多进程的环境下被其他进程的操作污染，整体是类似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后在写入，读取和写入应为一体的；

原子性实行函数Atomic的实现如下：

/ 原子操作 - 无法对锁本身进行原子性操作 - 只担保handler是否被原子性触发，对其逻辑是否抛出非常不卖力 - 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;} // 创建锁 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 { // 创建锁 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循环来进行壅塞抢占，为了不将当提高程壅塞去世，我们还加入了一个保险，由self::$fuse供应；

这里在实践过程中须要把稳的是，Atomic在传入回调函数时切勿再利用匿名函数作为参数值或者是通过use传入一个匿名函数，如:

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

APCu底层会对函数参数值或引用参数进行序列化储存，但匿名函数不可以被序列化，以是会抛出一个非常；但你可以通过当前工具的属性值或者静态属性来保存一个匿名函数，然后在Atomic的回调内调用利用。

由于目前我利用Webman基于SQLite和共享内存在自行实现一个具备RAFT的轻调度做事插件和做事注册与创造插件，以是特此为其完善增加了Channel特性；

Channel可以赞助实现类似Redis-List、Redis-stream、Redis-Pub/Sub的功能。

Channel

Channel是个分外的数据格式，他的格式是固定如下的：

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

它在共享内存中的键默认以#Channel#开头。

实现一个List

由于监听器创建消费是基于workerId的，我们可以通过不同进程创建相同的workerId的监听器来对同一个子通道进行监听：

A进程利用list作为workerId：

Cache::ChCreateListener('test', 'list', function(string $channelKey, string|int $workerId, mixed $message) {// TODO 你的业务逻辑});

B进程也同样创建list的workerId监听器：

Cache::ChCreateListener('test', 'list', function(string $channelKey, string|int $workerId, mixed $message) {// TODO 你的业务逻辑});

此时Channel test的数据如下：

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

把稳：共享内存中储存的futureId为末了一个监听器创建的futureId；当当提高程须要对监听器进行移除时，请勿利用该数据，对应进程内可以通过Cache::ChCreateListener()的返回值获取到当提高程创建的futureId用于移除监听器，不该用共享内存中储存的futureId即可

实现一个Pub/Sub

A进程利用workerman的workerId作为workerId：

Cache::ChCreateListener('test', $worker->id, function(string $channelKey, string|int $workerId, mixed $message) {// TODO 你的业务逻辑});

B进程利用workerman的workerId作为workerId：

Cache::ChCreateListener('test', $worker->id, function(string $channelKey, string|int $workerId, mixed $message) {// TODO 你的业务逻辑});

此时Channel test的数据可能如下：

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

这时，任意进程通过Cache::ChPublish('test', '这是一个测试'， false);发送即可。

注：发送第三个参数利用false时，如发送时还未创建监听器，则不会储存至Channel，即监听后才可存在

与Pub/Sub相同，只不过发布利用Cache::ChPublish('test', '这是一个测试'， true);， 当发布指定workerId时，可以实现类似Redis-Stream Group的功能。

注：这里更繁芜的功能可能须要对workerId进行变通，不能大略利用workerman自带的workerId，只须要自行方案好即可