book_admin/application/admin/command/RefreshBookShelf.php

<?php

namespace app\admin\command;

use app\common\constants\Common;
use app\common\library\Redis;
use think\Config;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\Output;
use think\Db;
use think\Log;
use think\Request;

class RefreshBookShelf extends Command
{
    private $dbConnects = [];
    private $_cacheBookIds = [];
    private $_cacheChapterIds = [];
    private $insertShelfData = [];
    private $_bookAutoShelfChapterNum = 0;
    private $recommendBookIds = [];


    private $getChapterIdsCount = 0;
    private $insertShelfCount = 0;
    private $getUserBookShelfListCount = 0;

    protected function configure()
    {
        $this->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 = <<<SQL
select id, book_id, user_id, chapter_id from user_recently_read where flag = 1 and id>$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 = <<<SQL
insert into book_shelf (user_id, book_id, insert_type, chapter_id, idx, createtime, updatetime) values 

SQL;

        $count = count($this->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']][] = <<<SQL
({$item['user_id']}, {$item['book_id']}, 2, {$item['chapter_id']}, 0, {$time}, {$time})

SQL;
                    }
                    $subTime = time() - $begTime;
                    Log::info('charge_in_time:' . $subTime . 's');

                    foreach ($insertGroupList as $dbName => $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 = <<<SQL
select user_id, book_id from book_shelf where user_id in (%s)
SQL;
            $strUserIds = implode(',', $_userIds);
            $sql = sprintf($sql, $strUserIds);
            $bookResult = $db->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));
    }

}