video_server/app/Jobs/VideoViewedJob.php

<?php

namespace App\Jobs;

use App\Model\Mongo\VideoHistoryModel;
use App\Model\VideoModel;
use App\Repositories\UserRepository;
use Carbon\Carbon;
use Hyperf\AsyncQueue\Job;
use Modules\Api\Services\VideoService;

class VideoViewedJob extends Job
{
    public $params;

    public function __construct($params)
    {
        // 这里最好是普通数据，不要使用携带 IO 的对象，比如 PDO 对象
        $this->params = $params;
    }

    public function handle()
    {
        $arrUserCodes = [];
        $nowDate = Carbon::now()->toDateString();
        $videoService = make(VideoService::class);
        if (!empty($this->params)) {
            foreach ($this->params as $item) {
                $data = [
                    'date' => $item['date'],
                    'user_id' => $item['user_id'],
                    'video_id' => $item['video_id'],
                ];
                $key = $data['date'] . '_'. $data['user_id']. '_'. $data['video_id'];
                $_id = substr(md5($key), 8, 16);
                $data['_id'] = $_id;
                $count = make(VideoHistoryModel::class)->db()->countDocuments(
                    ['_id'  => $_id]
                );
                if ($count) {
                    continue;
                }
                // 自动增加观看次数.
                $videoObj = VideoModel::where(['id' => $item['video_id']])->first(['id', 'watch', 'real_watch']);
                if (!empty($videoObj)) {
                    $videoObj->watch++;
                    $videoObj->real_watch++;
                    $videoObj->save();
                }
//                $userDayCount = make(VideoHistoryModel::class)->db()->countDocuments(
//                    ['uid'  => $item['user_id'],'date' => $data['date']]
//                );
                // 不管是不是会员，超过每日观影次数没有，都记录.
                $videoService->addHistory($data);
                if (!isset($arrUserCodes[$item['user_code']])) {
                    $arrUserCodes[$item['user_code']] = $item['user_code'];
                }
            }
        }

        if (!empty($arrUserCodes)) {
            $userRepo = make(UserRepository::class);
            foreach ($arrUserCodes as $k => $userCode) {
                // 清理用户缓存.
                $userRepo->delUserCache($userCode);
            }
        }
    }

}