2

我们有多个任务要在同一个 api 上运行,因为 api 是速率限制的,我们需要限制调用的次数(100 x 120 秒),这样我们就不会被锁定。每个锁定持续 30 分钟,会冻结整个团队并可能产生可怕的问题。

系统建立在 Laravel 之上,我们使用队列来进行这个 api 调用。我们需要的第一项工作是,系统每晚都会对 api 进行大约 950 次调用,以将 api 上的内容与我们的数据库同步。第二个和第三个作业将是按需的,它可能需要在用户需要时分别运行大约 100 到 200 个调用。(多个用户可能在一天中的同一时间需要这个,所以队列应该防止我们超过速率限制)。

我们在主管中设置了所有内容,只有一名工人,而且似乎工作正常。但后来我们意识到它并没有按需调用所有作业(第 2 和第 3 类),如果用户需要对 8 件事情采取行动,他们会出现在队列中,但只运行 2 或 3。其余的会得到永远不要拿起或不运行,甚至没有失败。通过将工人数量增加到 8 个,这在一定程度上得到了解决。在我们这样做之后,如果用户需要对 8 个工作采取行动,但我们每晚的 900 个任务开始失败。

我们还注意到它可以处理 8 个工作,但如果我们让用户移动多达 50 个任务,它也会失败。

我们在 /etc/supervisor/conf.d 上的主管配置:

    [unix_http_server]
    file = /tmp/supervisor.sock
    chmod = 0777
    chown= ubuntu:ubuntu

    [supervisord]
    logfile = /home/ubuntu/otmas/supervisord.log
    logfile_maxbytes = 50MB
    logfile_backups=10
    loglevel = info
    pidfile = /tmp/supervisord.pid
    nodaemon = false
    minfds = 1024
    minprocs = 200
    umask = 022
    user = ubuntu
    identifier = supervisor
    directory = /tmp
    nocleanup = true
    childlogdir = /tmp
    strip_ansi = false

    [rpcinterface:supervisor]
    supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface

    [supervisorctl]
    serverurl = unix:///tmp/supervisor.sock

    [program:laravel-worker]
    process_name=%(program_name)s_%(process_num)02d
    command=php /home/ubuntu/www/artisan horizon
    autostart=true
    autorestart=true
    user=ubuntu
    numprocs=8
    redirect_stderr=true
    stdout_logfile=/home/ubuntu/otmas/worker.log

我们的队列配置:

'connections' => [

        'redis' => [
            'driver' => 'redis',
            'connection' => 'default',
            'queue' => 'default',
            'retry_after' => 460,
            'block_for' => 140,
        ],

    ],

我们的地平线配置:

    'environments' => [
        'production' => [
            'supervisor-1' => [
                'connection' => 'redis',
                'queue' => ['default'],
                'balance' => 'simple',
                'processes' => 10,
                'tries' => 3,
            ],
        ],

        'local' => [
            'supervisor-1' => [
                'connection' => 'redis',
                'queue' => ['default'],
                'balance' => 'simple',
                'processes' => 3,
                'tries' => 3,
            ],
        ],

        'develop' => [
            'supervisor-1' => [
                'connection' => 'redis',
                'queue' => ['default','email'],
                'balance' => 'auto',
                'processes' => 3,
                'tries' => 3,
                'timeout' => 3,
                'delay' => 3,
            ],
        ],
    ],

我们在 database.php 上的 redis 配置:

    'redis' => [

        'client' => 'predis',

        'default' => [
            'host' => env('REDIS_HOST', '127.0.0.1'),
            'password' => env('REDIS_PASSWORD', null),
            'port' => env('REDIS_PORT', 6379),
            'database' => env('REDIS_DB', 0),
            'read_write_timeout' => -1,
        ],

        'cache' => [
            'host' => env('REDIS_HOST', '127.0.0.1'),
            'password' => env('REDIS_PASSWORD', null),
            'port' => env('REDIS_PORT', 6379),
            'database' => env('REDIS_CACHE_DB', 1),
            'read_write_timeout' => -1,
        ],

    ],

夜间作业:

class GetProjectTasks implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $project;
    public $tries = 540;
    public $timeout = 15;

    public function __construct(ZohoProject $project)
    {
        $this->project = $project;
    }

    public function handle()
    {
        \Log::info("before redis get tasks for project: ".$this->project->id);
        try {
            $client = new GuzzleClient();
            $accessToken = Setting::where('name','zoho_access_token')->get()->first();
            $requestHeaders = [
                    'cache-control' => 'no-cache',
                    'Authorization' => 'Bearer '.$accessToken->value
                ];
            $requestOptions = [
                'headers' => $requestHeaders
            ];
            $zohoTasksRestApi = Setting::where('name','zoho_projectapi_restapi')->get()->first()->value;
            $project = $this->project->id;
        } catch(Exception $e) {
            Log::notice('Exception caught');
            throw $e;
        }
        Redis::throttle('zohoprojecttasksget')->allow(85)->every(120)->then(function () use ($client, $zohoTasksRestApi, $portalId, $project, $requestOptions) {
            try {
                $zohoTasksResponse = $client->request('GET', $zohoTasksRestApi.'portal/'.$portalId.'/projects/'.$project.'/tasks/', $requestOptions);
                $zohoTasksResult = json_decode((string) $zohoTasksResponse->getBody(), true);
                if($zohoTasksResult) {
                    foreach ($zohoTasksResult['tasks'] as $key => $task) {
                        if (array_has($task, 'start_date') && array_has($task, 'end_date')) {
                            $taskStartDate = \Carbon\Carbon::createFromTimestamp((int)$task['start_date_long']/1000, 'America/Santiago');
                            $taskEndDate = \Carbon\Carbon::createFromTimestamp((int)$task['end_date_long']/1000, 'America/Santiago');
                            $zohoTask = ZohoTask::updateOrCreate(['id' => $task['id']], [
                                'name' => $task['name'],
                                'zoho_project_id' => $project,
                                'registry'  => json_encode($task),
                                'start_date' => $taskStartDate,
                                'end_date' => $taskEndDate
                            ]);
                            if(array_has($task, ['details.owners'])) {
                                if($task['details']['owners'][0]['name'] != 'Unassigned') {
                                    foreach ($task['details']['owners'] as $key => $owner) {
                                        $user = ZohoUser::find($owner['id']);
                                        $zohoTask->assignees()->save($user);
                                    }
                                }
                            }
                        } else {
                            $zohoTask = ZohoTask::updateOrCreate(['id' => $task['id']], [
                                'name' => $task['name'],
                                'zoho_project_id' => $project,
                                'registry'  => json_encode($task),
                            ]);
                        }
                    }
                }
            } catch(Exception $e) {
                Log::notice('Exception caught');
                throw $e;
            }
        }, function () use ($project) {
            return $this->release(85);
        });
    }
}

按需作业示例:

class ReplaceTask implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $task, $taskToReplace;
    public $tries = 140;
    public $timeout = 15;

    public function __construct(ZohoTask $task, ZohoTask $taskToReplace)
    {
        $this->task = $task;
        $this->taskToReplace = $taskToReplace;
    }

    public function handle()
    {
        \Log::info("before redis replace task: ".$this->taskToReplace.' with task:: '.$this->task);
        try {
            $client = new GuzzleClient();
            $accessToken = Setting::where('name','zoho_access_token')->get()->first();
            $requestHeaders = [
                    'cache-control' => 'no-cache',
                    'Authorization' => 'Bearer '.$accessToken->value
                ];
            $ZohoProjectsRestApi = Setting::where('name','zoho_projectapi_restapi')->get()->first()->value;
            $task = $this->task;
            $taskToReplace = $this->taskToReplace;
            $project = $task->project;
        } catch(Exception $e) {
            Log::notice('Exception caught');
            throw $e;
        }
        \Log::info("before redis task: ".$task->id.' to replace '.$taskToReplace->id.' Project: '.$project->id);
        Redis::throttle('zohoupdatetask')->allow(50)->every(120)->then(function () use ($client, $ZohoProjectsRestApi, $portalId, $project, $requestHeaders, $task, $taskToReplace) {
                try {
                    $assignee = $taskToReplace->assignees->last();
                    $assignResponse = $client->request('POST', $ZohoProjectsRestApi.'portal/'.$portalId.'/projects/'.$project->id.'/tasks/'.$task->id.'/', [
                          'headers' => $requestHeaders,
                          'form_params' => [
                            'start_date' => $taskToReplace->start_date->format('m-d-Y'),
                            'start_time' => $taskToReplace->start_date->format('h:i A'),
                            'person_responsible' => $assignee->id
                          ]
                      ]);
                    \Log::info($assignResponse->getBody());
                    if($assignResponse->getStatusCode() == 200) {
                        $task->assignees()->detach();
                        $assignResult = json_decode((string) $assignResponse->getBody(), true);
                        $taskRegistry = $assignResult['tasks'][0];
                        $taskStartDate = \Carbon\Carbon::createFromTimestamp((int)$taskRegistry['start_date_long']/1000, 'America/Santiago');
                        $taskEndDate = \Carbon\Carbon::createFromTimestamp((int)$taskRegistry['end_date_long']/1000, 'America/Santiago');
                        $task->start_date = $taskStartDate;
                        $task->end_date = $taskEndDate;
                        $task->registry = json_encode($taskRegistry);
                        $task->assignees()->save($assignee);
                        $task->save();
                        \Log::info("Task Ok move");
                    }
                } catch(Exception $e) {
                    Log::notice('Exception caught');
                    throw $e;
                }
            }, function () use ($project, $task, $taskToReplace) {
                \Log::info("Task hit throttle");
            return $this->release(120);
        });
        \Log::info("after redis");
    }
}

地平线任务:(被搁置的任务永远不会运行或拿起或任何东西) 任务出现在队列中,但似乎永远不会运行或失败

4

0 回答 0