多进程代码分析

问答 / 1849 / 1 / 创建于 6年前

老大让我拉取公众号的openid,大约1000万条，我写了个daemon,单进程，拉取速度太慢，要跑4天，老大用多进程写了一个，只要跑6个小时，但是他的代码我看不懂，大神能不能帮我分析下解释下，我关键是他的第0个任务分配任务看不得不是特别的明白下面粘贴他的代码

<?php

namespace App\Console\Commands\MutualPlan;

use Illuminate\Console\Command;
use App\Models\MutualOauth;
use App\Models\Oauth;
use App\Enums\Redis;
use App\Models\MutualOpenidList;

class GetOpenId extends Command
{
/**

The name and signature of the console command.
@var string
*/
protected $signature = 'mutual-plan:get-openid {process}';

/**
The console command description.
@var string
*/
protected $description = '获取微信公众号的OpenID';

/**
next_openid
*/
protected $next_openid;

protected $openid;

protected $last_insert_id;

protected $process;

/**
Create a new command instance.

@return void
*/
public function construct()
{
parent::construct();

$this->redis = getRedis('cache_redis', ['prefix' => Redis\DBPREFIX::HUZHU_CACHE_REDIS]);

}

/**

Execute the console command.

@return mixed
*/
public function handle()
{

if (false) {
    $keys = $this->redis->keys('getopenid_ll_nextopendid_*');

    foreach ($keys as $key => $item) {
        dump($item, $this->redis->del(str_replace('huzhu.com:cache:', '', $item)));
    }

    exit;
}

$this->process = $this->argument('process');

$timestamp = time();

if ($this->process == '0') {

    $index = 0;

    while (true) {

        if ((time() - $timestamp) > '600') {

            sleep(10);

            exit;
        }

        if ($index >= '9') {
            $index = 1;
        } else {
            $index++;
        }

        $currentOpenid = $this->redis->get('getopenid_ll_nextopendid_' . $index);

        if (!$currentOpenid) {
            $next = $this->redis->get('wechat-all-openid-job-list');

            $access_token_url = sprintf('https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s', env('WECHAT_MIC_APPID'), env('WECHAT_MIC_SECRET'));

            $access_token_info = \GuzzleHttp\json_decode(file_get_contents($access_token_url));

            $access_token = $access_token_info->access_token;

            if (empty($next)) {

                $openid_url = sprintf('https://api.weixin.qq.com/cgi-bin/user/get?access_token=%s', $access_token);

                $openid_info = \GuzzleHttp\json_decode(file_get_contents($openid_url));

                $this->redis->set('getopenid_ll_nextopendid_' . $index, 'empty');

            } else {
                $openid_url = sprintf('https://api.weixin.qq.com/cgi-bin/user/get?access_token=%s&next_openid=%s', $access_token, $next);

                $openid_info = \GuzzleHttp\json_decode(file_get_contents($openid_url));

                $this->redis->set('getopenid_ll_nextopendid_'  . $index, $openid_info->next_openid,3600*24*5);
            }

            $this->info('make-job-' . $index . '-' . $openid_info->next_openid);

            $this->redis->set('wechat-all-openid-job-list', $openid_info->next_openid, 86400);

        }

        continue;

    }

}

$job = $this->redis->get('getopenid_ll_nextopendid_' . $this->process);

if ($job) {
    $this->warn('handle-job-' . $job);

    $this->redis->del('getopenid_ll_nextopendid_' . $this->process);

    $this->runWechatOpenid($job);
}

sleep(60);

exit();

}

private function runWechatOpenid($openid)
{
try {
$access_token_url = sprintf('https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s', env('WECHAT_MIC_APPID'), env('WECHAT_MIC_SECRET'));

    $access_token_info = \GuzzleHttp\json_decode(file_get_contents($access_token_url));

    $access_token = $access_token_info->access_token;

    if ($openid == 'empty') {

        $openid_url = sprintf('https://api.weixin.qq.com/cgi-bin/user/get?access_token=%s', $access_token);

    } else {

        $openid_url = sprintf('https://api.weixin.qq.com/cgi-bin/user/get?access_token=%s&next_openid=%s', $access_token, $openid);

    }

    $openid_info = \GuzzleHttp\json_decode(file_get_contents($openid_url));

    $openids = $openid_info->data->openid;

    if (empty($openids) and empty($openid_info->next_openid)) {

        \Log::info($this->description.'为空');

        sleep(10);

        exit;
    }

    foreach ($openids as $openid) {

        $this->openid = $openid;

        /**
         * 判断此openid 是否存在表里  没有插入
         */
        $is_oauth = Oauth::where('authType','203')->where('openid',$this->openid)->first();

        if ($is_oauth) {
            continue;
        }

        $is_mutualoauth = MutualOauth::where('auth_type','1')->where('open_id',$this->openid)->first();

        if ($is_mutualoauth) {
            continue;
        }

        $this->info('save-' . $openid);

        $insert = new \App\Models\MutualOpenidList();
        $insert->openid = $this->openid;

        if ($insert->save() === true) {

            $this->last_insert_id = $insert->id;
        }
    }

} catch (\Exception $exception) {

    \Log::error($this->description . '异常',
        [
            'php内存使用量' => bcdiv(memory_get_usage(), bcmul(1024,1024,2)) . 'M',
            '插入openid' => $this->openid,
            'msg' => $exception->getMessage()
        ]
    );

}

$this->error('done');

exit;

}
}