setName('RefreshBookShelf') ->addArgument('startDbNum', Argument::REQUIRED, "第几个数据库开始") ->addArgument('endDbNum', Argument::REQUIRED, "第几个数据库结束") ->setDescription('刷新用户书架，用户阅读超过第十章的书籍加入书架'); } protected function execute(Input $input, Output $output) { $beginTime = time(); Request::instance()->module('admin'); $startDbNum = $input->getArgument('startDbNum'); $endDbNum = $input->getArgument('endDbNum'); $output->writeln('开始数据库' . $startDbNum); $output->writeln('结束数据库' . $endDbNum); if ($startDbNum > 511 || $endDbNum > 511) { $output->writeln('参数错误，数据库开始序号和结束序号不能大于511'); die(); } $model = model('BookshelfRecommand'); $this->recommendBookIds = $model->where('sex', 1)->where('status', Common::STATUS_NORMAL)->limit(6)->column('book_id'); $this->_bookAutoShelfChapterNum = 8; Log::info("开始刷书架，开始时间:" . date("Y-m-d H:i:s", time())); for ($idx = $startDbNum; $idx <= $endDbNum; $idx++) { try { $output->writeln($idx); Log::info("***************************************************************************"); $db = $this->dbConnect($idx, 'shard'); $userRecentlyReadId = 0; while (true) { Log::info(sprintf('dbNo:%s,begin_id：%s', $idx, $userRecentlyReadId)); $sql = <<$userRecentlyReadId limit 10000; SQL; $result = $db->query($sql); if ($result) { foreach ($result as $item) { $autoShelf = $this->checkChapterIdAutoShelf($item['book_id'], $item['chapter_id']); if ($autoShelf) { $this->preAutoBookShelf($item['user_id'], $item['book_id'], $item['chapter_id']); } } $lastItem = end($result); $userRecentlyReadId = $lastItem['id']; $this->autoBookSelf(false); Log::info(sprintf('dbNo:%s,last_id：%s', $idx, $userRecentlyReadId)); $this->cleanCount(); } else { $this->autoBookSelf(true); Log::info(sprintf('finish_dbNo：%s,last_id：%s', $idx, $userRecentlyReadId)); $this->cleanCount(); break; } sleep(0.05); } $this->closeDbConnect(); } catch (\Exception $e) { Log::error($e->getMessage() . $e->getTraceAsString()); Log::error("数据库" . $db->getConfig('database') . "中的信息，出错了,需要重新执行"); } } $endTime = time(); $usedTime = $endTime - $beginTime; Log::info("刷书架完成，结束时间:" . date("Y-m-d H:i:s", time()) . " 耗时：" . $usedTime . 's'); } /** * 获取数据库连接 * @param $param 编号 * @param string $deploy 业务 * @return \think\db\Connection * @throws \think\Exception */ private function dbConnect($param, $deploy) { $db_config = $this->get_db_deploy($param, $deploy); if (empty($this->dbConnects[$db_config['database']])) { Log::info(sprintf('打开数据库连接，database:%s', $db_config['database'])); $this->dbConnects[$db_config['database']] = Db::connect($db_config); } return $this->dbConnects[$db_config['database']]; } /** * 关闭数据库连接 */ private function closeDbConnect() { foreach ($this->dbConnects as $database => $dbConnect) { Log::info(sprintf('关闭数据库连接，database:%s', $database)); $dbConnect->close(); } $this->dbConnects = []; } /** * 获取db分库的配置参数 * * @param string|int $param 取模值 * @param string $deploy 分库前缀 * @return array */ function get_db_deploy($param, $deploy = 'shard') { $db = Config::get('db'); $mod = $param % $db[$deploy . '_num']; $mod = abs($mod); $list = explode(';', $db[$deploy . '_list']); foreach ($list as $item) { $con = explode(':', $item); // 0=0-191库编号 1=192.168.1.149主IP 2=3306主端口 3=192.168.1.150从IP 4=3306从端口 if (count($con) >= 3) { $c = explode('-', $con[0]); //库编号 0开始 1结束 if (count($c) >= 2) { if ($c[0] <= $mod && $mod <= $c[1]) { $database = Config::get('database'); if ($database['deploy'] == 1 && count($con) >= 5) { //开启主从 & 带主从配置 $database['hostname'] = $con[1] . ',' . $con[3]; //192.168.1.149,192.168.1.150 $database['hostport'] = $con[2] . ',' . $con[4]; //3306,3306 } else { //只有主库 $database['hostname'] = $con[1]; $database['hostport'] = $con[2]; } $database['database'] = str_replace('$mod', $mod, $db[$deploy . '_database']); return $database; } } } } Log::error("分库获取失败！"); return []; } /** * 检查当前章节是否符合自动加入书架的条件 * @param $bookId 书籍id * @param $chapterId 章节id * @return bool */ private function checkChapterIdAutoShelf($bookId, $chapterId) { $result = false; if (in_array($bookId, $this->_cacheBookIds, true)) { if (!in_array($chapterId, $this->_cacheChapterIds, true)) { Log::info("putIn,bId:$bookId,cId:$chapterId"); $result = true; } } else { $chapterIds = $this->getChapterIds($bookId); $this->_cacheBookIds[] = $bookId; $this->_cacheChapterIds = array_merge($this->_cacheChapterIds, $chapterIds); if (!in_array($chapterId, $chapterIds, true)) { Log::info("doNothing1,bId:$bookId,cId:$chapterId"); $result = true; } } if (!$result) { Log::info("doNothing2,bId:$bookId,cId:$chapterId"); } if (count($this->_cacheBookIds) > 10000 || count($this->_cacheChapterIds) > 30000) { $this->_cacheBookIds = null; $this->_cacheChapterIds = null; Log::info('cl_cch'); } return $result; } /** * 获取书籍章节id列表 * @param $bookId * @return array */ private function getChapterIds($bookId) { $result = []; if (Config::get('redis.change') == 1) { $code = $this->getCode('A04' . $bookId); $redis = Redis::instanceBookChange($code); $list = $redis->zRange('A04' . $bookId, 0, $this->_bookAutoShelfChapterNum); } else { $redis = Redis::instanceBook(); $list = $redis->lRange("basedata_chapter_owchcp_{$bookId}", 0, $this->_bookAutoShelfChapterNum); } foreach ($list as $key => $value) { if (Config::get('redis.change') == 1) { $json = json_decode($value, true); } else { $json = json_decode(gzdecode($value), true); } $chapterId = $json[4] ?? $json['chapterId']; $result[] = intval($chapterId); } $this->getChapterIdsCount++; return $result; } private function getCode($val) { $bKey = md5($val, true); $rv = (ord($bKey[3]) & 0xFF) << 24 | (ord($bKey[2]) & 0xFF) << 16 | (ord($bKey[1]) & 0xFF) << 8 | ord($bKey[0]) & 0xFF; return $rv & 0xffffffff; } private function preAutoBookShelf($userId, $bookId, $chapterId) { $this->insertShelfData[] = [ 'user_id' => $userId, 'book_id' => $bookId, 'chapter_id' => $chapterId, ]; } /** * 将缓存的书架信息插入数据库 * @param bool $forceInsert 强制将缓存的数据插入数据库 * @throws \think\Exception * @throws \think\exception\PDOException */ private function autoBookSelf($forceInsert = false) { $time = time(); $sql = <<insertShelfData);//总条数 if ($count > 100000 || $forceInsert) { $page = 0; $pageSize = 10000; while (true) { $start = $page * $pageSize; $_insertShelfDataPart = array_slice($this->insertShelfData, $start, $pageSize); if (empty($_insertShelfDataPart)) { $this->insertShelfData = null; break; } else { $insertGroupList = []; $userIds = array_column($_insertShelfDataPart, 'user_id'); $userIdBookIdRels = $this->getUserBookShelfList($userIds); $begTime = time(); foreach ($_insertShelfDataPart as $item) { $_userIdBookId = sprintf('%s_%s', $item['user_id'], $item['book_id']); if (in_array($_userIdBookId, $userIdBookIdRels, true)) { Log::info(sprintf('doNothing3,uId:%s,bId:%s', $item['user_id'], $item['book_id'])); continue; } $config = $this->get_db_deploy($item['user_id'], 'shelf'); $insertGroupList[$config['database']][] = << $values) { $aas = explode('_', $dbName); if (count($aas) > 0) { $param = end($aas); $insValStr = implode(',', $values); $sqlInsertBat = $sql . $insValStr; $db = $this->dbConnect($param, 'shelf'); $db->execute($sqlInsertBat); $this->insertShelfCount++; } } } sleep(0.05); $page++; } $this->insertShelfData = []; } } /** * 获取已经存在的用户书架关系 * @param $userIds * @return array * @throws \think\Exception * @throws \think\exception\PDOException */ private function getUserBookShelfList($userIds) { $begTime = time(); $result = []; $userGroupList = []; foreach ($userIds as $userId) { $config = $this->get_db_deploy($userId, 'shelf'); $userGroupList[$config['database']][] = $userId; } foreach ($userGroupList as $dbName => $_userIds) { $_userId = end($_userIds); $db = $this->dbConnect($_userId, 'shelf'); $sql = <<query($sql); $this->getUserBookShelfListCount++; foreach ($bookResult as $item) { $result[] = sprintf('%s_%s', $item['user_id'], $item['book_id']); } } $subTime = time() - $begTime; Log::info('getUserBookShelfList:' . $subTime); return $result; } private function cleanCount() { Log::info(sprintf('getChapterIdsCount:%s,insertShelfCount:%s,getUserBookShelfListCount:%s', $this->getChapterIdsCount, $this->insertShelfCount, $this->getUserBookShelfListCount)); } }