删除任务

@param string   $queue
@param RedisJob $job
@return void
340115
public function deleteReserved($queue, $job)
    {
        $this->redis->zRem($this->getQueue($queue) . ':reserved', $job->getReservedJob());
    }
230383
protected function fail($job)
    {
        Queue::getFailLogger()->log($job->getQueue()->getConnectionName(), $job->getQueueName(), $job->getRawBody());
        $job->delete();
    }
684647
public function set(string $data, string $queue = 'default')
    {
        return $this->redis->rPush($queue, $data);
    }
1129326
public function delete(Job $job)
    {
        $result = false;
        try {
            $res = $this->_queue->deleteMessage($job->getInstance()->getReceiptHandle());
            $result = $res->isSucceed();
        } catch (\Exception $e) {
            $this->eventsManager->fire('listener:logger', $this, '删除Queue消息失败:' . $e->getMessage());
        } finally {
            return $result;
        }
    }
1273044
public function handle(Queue $queue, JobRepository $jobs)
    {
        if (is_null($job = $jobs->findFailed($this->id))) {
            return;
        }

        $queue->connection($job->connection)->pushRaw(
            $this->preparePayload($id = JobId::generate(), $job->payload), $job->queue
        );

        $jobs->storeRetryReference($this->id, $id);
    }
1188309
public function delete()
    {
        parent::delete();

        $this->redis->deleteReserved($this->queue, $this->job);
    }
1074745
protected function prepareMessage($queue, Job $job)
    {
        $this->queueDeclare($queue);

        return new AMQPMessage(json_encode($job->createPayload()), $job->getOptions());
    }
1086842
public void queueJob(Runnable job) {
        try {
            logger.debug("Queueing job {}", job);
            executor.submit(job);
        } catch (RejectedExecutionException e) {
            logger.error("Unable to queue a send-command-job! ", e);
        }
        checkQueueSize();
    }
1157496
private function runJob($job)
    {
        try {
            $this->logger->notice("Job #{$this->currentJobId} [{$job->getName()}] started.");
            $status = $this->broker->getStatus($job->getId());
            $status['status'] = Job::STATUS_RUNNING;
            $status['started'] = date(DATE_RFC3339_EXTENDED);
            $this->broker->setStatus($job->getId(), $status);
            $job->setup();
            $response = $job->go();
            $job->tearDown();
            $status['status'] = Job::STATUS_FINISHED;
            $status['finished'] = date(DATE_RFC3339_EXTENDED);
            $status['response'] = $response;
            $this->broker->setStatus($job->getId(), $status);
            $this->logger->notice("Job #{$this->currentJobId} [{$job->getName()}] finished.");
        } catch (\Exception $e) {
            $status['status'] = Job::STATUS_FAILED;
            $status['failed'] = date(DATE_RFC3339_EXTENDED);
            $this->broker->setStatus($job->getId(), $status);
            $this->logException($e, "Job #{$this->currentJobId}  [{$job->getName()}] Exception");
            $this->logger->alert("Job #{$this->currentJobId}  [{$job->getName()}] died.");
        }
    }
1010220
public function deleteReserved($queue, $job)
    {
        parent::deleteReserved($queue, $job);

        $this->event($this->getQueue($queue), new JobDeleted($job, $job->getReservedJob()));
    }
472284
public function push($job, $data = '', $queue = null)
	{
		$queueJob = $this->resolveJob($this->createPayload($job, $data, $queue));

		try
		{
			$queueJob->fire();
		}
		catch(Exception $e)
		{
			$this->handleFailedJob($queueJob);

			throw $e;
		}

		return 0;
	}
760082
public function deleteAndRelease($queue, $job, $delay)
    {
        $queue = $this->getQueue($queue);

        $this->getConnection()->eval(
            LuaScripts::release(), 2, $queue.':delayed', $queue.':reserved',
            $job->getReservedJob(), $this->availableAt($delay)
        );
    }
302079
public function push($job, $data = '', $queue = null)
    {
        $queueJob = $this->resolveJob($this->createPayload($job, $queue, $data), $queue);

        try {
            $this->raiseBeforeJobEvent($queueJob);

            $queueJob->fire();

            $this->raiseAfterJobEvent($queueJob);
        } catch (Exception $e) {
            $this->handleException($queueJob, $e);
        } catch (Throwable $e) {
            $this->handleException($queueJob, new FatalThrowableError($e));
        }

        return 0;
    }
268763
protected function createPayloadArray($job, $queue, $data = '')
    {
        return array_merge(parent::createPayloadArray($job, $queue, $data), [
            'queue' => $this->getQueue($queue),
        ]);
    }
1086487
public function executeJob(Job $job, $maxRetries = 1)
    {
        $this->dispatcher->dispatch(WorkerEvents::EXECUTE_JOB, new JobEvent($job));

        $stats    = $this->pheanstalk->statsJob($job);
        $payload  = (array) json_decode($job->getData(), true);
        $releases = intval($stats['releases']);
        $priority = intval($stats['pri']);

        // context for logging
        $context = [
            'tube'    => $stats['tube'],
            'payload' => $payload,
            'attempt' => $releases + 1,
        ];

        try {
            // execute command
            $result = $this->execute($stats['tube'], $payload);

            // delete job if it completed without exceptions
            $this->delete($job);

            return $result;
        } catch (RescheduleException $re) {
            // Override priority if the RescheduleException provides a new one.
            if (!is_null($re->getReshedulePriority())) {
                $priority = $re->getReshedulePriority();
            }
            // reschedule the job
            $this->reschedule($job, $re->getRescheduleDate(), $priority);
        } catch (AbortException $e) {
            // abort thrown from executor, rethrow it and let the caller handle it
            throw $e;
        } catch (\Throwable $e) {
            // some other exception occured
            $message = sprintf('Exception occurred: %s in %s on line %d', $e->getMessage(), $e->getFile(), $e->getLine());
            $this->logJob($job->getId(), $message, LogLevel::ERROR, $context);
            $this->logJob($job->getId(), $e->getTraceAsString(), LogLevel::DEBUG, $context);

            // see if we have any retries left
            if ($releases > $maxRetries) {
                // no more retries, bury job for manual inspection
                $this->bury($job);

                $this->dispatcher->dispatch(WorkerEvents::JOB_BURIED_EVENT, new JobBuriedEvent($job, $e, $releases));
            } else {
                // try again, regardless of the error
                $this->reschedule($job, new \DateTime('+10 minutes'), $priority);
            }
        }

        return false;
    }
137612
public function push($job, $data = '', $queue = null)
    {
        $this->lastPushed = $job;

        return parent::push($job, $data, $queue);
    }
645278
public function queuePush(Job $job)
    {
        if ($this->isSingle()) {
            $queue_config = $this->getQueueConfig();
            if (isset($queue_config['driver'])) {
                /** @var BaseQueue $driver */
                $driver = $queue_config['driver'];
                $driver::component()->push($job);
            }
        }
    }