video_server/app/Jobs/VideoWatchedByBulkJob.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;
use Psr\SimpleCache\CacheInterface;

class VideoWatchedByBulkJob 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)) {
            $cacheRepo = make(CacheInterface::class);
            $cacheKey = 'video:watch:bulk2';
            $cacheData = $cacheRepo->get($cacheKey, []);
            if (!empty($cacheData)) {
                if (count($cacheData) > 100) {
                    $cacheRepo->delete($cacheKey);
                    // 自动增加观看次数.
                    foreach ($cacheData as $videoId => $incNum) {
                        $videoObj = VideoModel::where(['id' => $videoId])->first(['id', 'watch', 'real_watch']);
                        if (!empty($videoObj)) {
                            $videoObj->watch = $videoObj->watch + $incNum;
                            $videoObj->real_watch = $videoObj->real_watch + $incNum;
                            $videoObj->save();
                        }
                    }
                }
            }
            foreach ($this->params as $item) {
                if (isset($cacheData[$item['video_id']])) {
                    $cacheData[$item['video_id']] = $cacheData[$item['video_id']] + 1;
                } else {
                    $cacheData[$item['video_id']] = 1;
                }
            }
            $cacheRepo->forever($cacheKey, $cacheData);
        }
    }

}