book_admin/application/admin/command/RefreshRecommendBookShelf.php

<?php

namespace app\admin\command;

use app\common\constants\Common;
use app\common\library\Redis;
use app\main\helper\ArrayHelper;
use think\Config;
use think\console\Command;
use think\console\Input;
use think\console\input\Option;
use think\console\Output;
use think\Db;
use think\Log;
use think\Request;

class RefreshRecommendBookShelf extends Command
{

    private $recommendBookIds = [];
    private $dbConnects = [];
    private $insertShelfData = [];

    protected function configure()
    {
        $this->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 = <<<SQL
select id, sex from user where id > $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 = <<<SQL
select distinct user_id from book_shelf where user_id in($strUserIds);
SQL;
        $result = $db->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 = <<<SQL
insert into book_shelf (user_id, book_id, insert_type, createtime, updatetime) values
SQL;

        $count = count($this->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[] = <<<SQL
({$item['user_id']}, {$item['book_id']}, 1, {$time}, {$time})
SQL;
                    }
                    $insValStr = implode(',', $insertValueList);
                    $sqlInsertBat = $sql . $insValStr;
                    $db = $this->dbConnect($item['user_id'], 'shelf');
                    $db->execute($sqlInsertBat);
                    sleep(0.05);
                }
                $page++;
            }
        }

    }
}