前文连接, 讲了 es 是如何启动 swoole 服务的.
里面有一个工具类 TableManager. 这个类为了处理进程间数据共享. 是对 swoole_table 的一层封装
swoole_table 一个基于共享内存和锁实现的超高性能, 并发数据结构. 用于解决多进程 / 多线程数据共享和同步加锁问题.
TableManager 主要做了下面几件事
add 方法
如果 $list 数组中有这个表名 ($name 是一个表名或者叫做集合名), 就初始化 swoole_table, 然后配置的字段类型数组进行创建
- if(!isset($this->list[$name])){
- $table = new Table($size);
- foreach ($columns as $column => $item){
- $table->column($column,$item['type'],$item['size']);
- }
- $table->create();
- $this->list[$name] = $table;
- }
get 方法
直接返回 swoole_table 的实例.
使用的地方有很多
前文提到的在系统设置 Cache 组件 Cache::getInstance() 的时候
构造方法做了如下事情
- $num = intval(Config::getInstance()->getConf("EASY_CACHE.PROCESS_NUM"));//Config 默认配置是 1, 如果配置为小于等于 0 则不开启 Cache
- if($num <= 0){
- return;
- }
- $this->cliTemp = new SplArray();
- // 若是在主服务创建, 而非单元测试调用
- if(ServerManager::getInstance()->getServer()){
- // 创建 table 用于数据传递
- TableManager::getInstance()->add(self::EXCHANGE_TABLE_NAME,[
- 'data'=>[
- 'type'=>Table::TYPE_STRING,
- 'size'=>10*1024
- ],
- 'microTime'=>[
- 'type'=>Table::TYPE_STRING,
- 'size'=>15
- ]
- ],2048);
- // 创建了一个__Cache 的 swoole_table 表, 字段为 data String 10240,microTime String 15 的表
- $this->processNum = $num;
- for ($i=0;$i <$num;$i++){
- ProcessManager::getInstance()->addProcess($this->generateProcessName($i),CacheProcess::class);
- }
- }
ProcessManager 也是一个很重要的概念. 其实就是一个管理任务映射的工具.
这里可以看到 ProcessManager::getInstance()->addProcess($this->generateProcessName($i),CacheProcess::class)
其实这里是通过 ProcessManager, 让 swoole 服务添加了一个进程. swoole 的 addProcess 方法, 文档链接 https://wiki.swoole.com/wiki/page/390.html
提前略带讲解一下 Cache 的 set 方法加深概念
- // 讲解一下 Cache 的 set 方法加深概念
- if(!ServerManager::getInstance()->isStart()){// 兼容测试模式. 也就是不开启服务的情景下直接是 clitemp 中取缓存数据
- $this->cliTemp->set($key,$data);
- }
- if(ServerManager::getInstance()->getServer()){
- $num = $this->keyToProcessNum($key);// 这里是通过 key 然后 hash 到应该投放的 Cache 进程中去.
- $msg = new Msg();
- $msg->setCommand('set');
- $msg->setArg('key',$key);
- $msg->setData($data);
- // 下面一句话还是挺复杂的, 根据 key 名 hash 到 ProcessManager 对应的映射, 然后获取到 swoole_process 的实例, 以 swoole 的 write 函数向管道内写入数据.
- ProcessManager::getInstance()->getProcessByName($this->generateProcessName($num))->getProcess()->write(\swoole_serialize::pack($msg));
- // 在写完数据后, 在 CacheProcess 的 onReceive 方法中可以看到对应 setCommand 的操作细节. 其实数据都被写到了一个 Arr 数组中. 下篇接着讲一下 Cache 的实现细节. 这节还是主要讲 TableManager 和它的相关作用.
- }
来源: https://www.cnblogs.com/gavinjunftd/p/9435702.html