setName('RefreshRecommendBookShelf') ->addOption('startDbNum', 's', Option::VALUE_REQUIRED, '第几个数据库开始') ->addOption('endDbNum', 'e', Option::VALUE_REQUIRED, '第几个数据库结束') ->addOption('channelIds', 'c', Option::VALUE_REQUIRED, '渠道商id列表,逗号分隔') ->setDescription('将书库推荐书插入用户书架'); } protected function execute(Input $input, Output $output) { $beginTime = time(); Request::instance()->module('admin'); //cli模式下无法获取到当前的项目模块,手动指定一下 $model = model('BookshelfRecommand'); $this->recommendBookIds[1] = $model->where('sex', 1)->where('status', Common::STATUS_NORMAL)->order('id')->limit(6)->column('book_id'); $this->recommendBookIds[2] = $model->where('sex', 2)->where('status', Common::STATUS_NORMAL)->order('id')->limit(6)->column('book_id'); $startDbNum = $input->getOption('startDbNum'); $endDbNum = $input->getOption('endDbNum'); $strChannelIds = $input->getOption('channelIds'); $output->writeln('开始数据库' . $startDbNum); $output->writeln('结束数据库' . $endDbNum); $output->writeln('渠道商id:' . $strChannelIds); if ($startDbNum > 511 || $endDbNum > 511) { $output->writeln('参数错误,数据库开始序号和结束序号不能大于511'); die(); } if (empty($strChannelIds)) { $output->writeln('渠道商id不能为空'); die(); } $channelIds = explode(',', $strChannelIds); foreach ($channelIds as $channelId) { if (!is_numeric($channelId)) { $output->writeln("$channelId 不是整数"); } else { $output->writeln($channelId); } } for ($idx = $startDbNum; $idx <= $endDbNum; $idx++) { try { $output->writeln($idx); Log::info("***************************************************************************"); $beginUserId = 0; while (true) { $result = $this->getUserInfo($idx, $beginUserId, $strChannelIds); if ($result) { $userIds = array_column($result, 'id'); $userIdsHasShelf = $this->getBookShelfUserIds($idx, $userIds); $userIds = array_diff($userIds, $userIdsHasShelf); $tmpRes = []; $bookShelfKeys = []; foreach ($userIds as $userId) { $tmpRes[] = ArrayHelper::array_column_search($result, 'id', $userId); $bookShelfKeys[] = "SBIK:$userId"; } $redis = Redis::instance(); $redis->del(...$bookShelfKeys); foreach ($tmpRes as $item) { $this->preAutoBookShelf($item['id'], $item['sex']); } $lastItem = end($result); $beginUserId = $lastItem['id']; $this->autoBookSelf(false); Log::info(sprintf('dbNo:%s,last_id:%s', $idx, $beginUserId)); sleep(0.05); } else { $this->autoBookSelf(true); Log::info(sprintf('finish_dbNo:%s,last_id:%s', $idx, $beginUserId)); break; } } $this->closeDbConnect(); } catch (\Exception $e) { Log::error($e->getMessage()); } } $endTime = time(); $usedTime = $endTime - $beginTime; Log::info("书库推荐书插入用户书架完成,结束时间:" . date("Y-m-d H:i:s", $endTime) . " 耗时:" . $usedTime . 's'); } /** * 获取用户信息 * @param $dbIdx 数据库索引 * @param $beginUserId 起始用户id * @param $strChannelIds 渠道商id列表 * @param int $limit * @return mixed * @throws \think\Exception * @throws \think\exception\PDOException */ private function getUserInfo($dbIdx, $beginUserId, $strChannelIds, $limit = 1000) { $db = $this->dbConnect($dbIdx, 'user'); Log::info(sprintf('dbNo:%s,begin_id:%s', $dbIdx, $beginUserId)); $sql = << $beginUserId and channel_id in($strChannelIds) limit $limit; SQL; $result = $db->query($sql); return $result; } /** * 获取有书架信息的用户id * @param $dbIdx 数据库索引 * @param array $userIds 用户id列表 * @return array|mixed * @throws \think\Exception * @throws \think\exception\PDOException */ private function getBookShelfUserIds($dbIdx, array $userIds) { if (empty($userIds)) { return []; } $strUserIds = implode(',', $userIds); $db = $this->dbConnect($dbIdx, 'shelf'); $sql = <<query($sql); $resUserIds = array_column($result, 'user_id'); return $resUserIds; } /** * 获取数据库连接 * @param $param 编号 * @param $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']])) { $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 $userId * @param $sex */ private function preAutoBookShelf($userId, $sex) { $sex = $sex == 0 ? 1 : $sex; $bookIds = $this->recommendBookIds[$sex]; foreach ($bookIds as $bookId) { $this->insertShelfData[] = [ 'user_id' => $userId, 'book_id' => $bookId, ]; } } /** * 将书架书籍插入数据库 * @param bool $forceInsert 是否强制刷入数据库。如果不强制刷入数据库,insertShelfData的数量大于10000会刷入数据库 * @throws \think\Exception * @throws \think\exception\PDOException */ private function autoBookSelf($forceInsert = false) { $time = time(); $sql = <<insertShelfData);//总条数 if ($count == 0) { return; } if ($count > 10000 || $forceInsert) { $page = 0; $pageSize = 1000; while (true) { $start = $page * $pageSize;//偏移量,当前页-1乘以每页显示条数 $_insertShelfDataPart = array_slice($this->insertShelfData, $start, $pageSize); if (empty($_insertShelfDataPart)) { $this->insertShelfData = null; break; } else { $insertValueList = []; foreach ($_insertShelfDataPart as $item) { $insertValueList[] = <<dbConnect($item['user_id'], 'shelf'); $db->execute($sqlInsertBat); sleep(0.05); } $page++; } } } }