入队.

@param string $data  需要放入队列的数据.
@param string $queue Tube名字,具体请参见beanstalk文档.
@param int    $delay 延迟ready的秒数,在这段时间job为delayed状态.默认不延迟

@return CTask
186908
public function set(string $data, string $queue = 'default', $delay = 0)
    {
        return $this->beanstalkTask->putInTube($queue, $data, null, $delay);
    }
818797
protected function createStringPayload($job, $queue, $data)
    {
        return $this->withCreatePayloadHooks($queue, [
            'displayName' => is_string($job) ? explode('@', $job)[0] : null,
            'job' => $job,
            'maxTries' => null,
            'delay' => null,
            'timeout' => null,
            'data' => $data,
        ]);
    }
873577
public function pushRaw($payload, $queue = null, array $options = array())
	{
		return $this->pheanstalk->useTube($this->getQueue($queue))->put($payload);
	}
684095
public function flush($delay = 0) {
    $result = $this->memcache->flush();
    $this->result_code = $result ? self::RES_SUCCESS : self::RES_NOTSTORED;
    return $result;
  }
398334
protected function pushToDatabase($delay, $queue, $payload, $attempts = 0)
	{
		$availableAt = $delay instanceof DateTime ? $delay : Carbon::now()->addSeconds($delay);

		return $this->database->table($this->table)->insertGetId([
			'queue' => $this->getQueue($queue),
			'payload' => $payload,
			'attempts' => $attempts,
			'reserved' => 0,
			'reserved_at' => null,
			'available_at' => $availableAt->getTimestamp(),
			'created_at' => $this->getTime(),
		]);
	}
1092309
public function queue($view, array $data, $callback, $queue = null)
    {
        $callback = $this->buildQueueCallable($callback);

        return $this->queue->push('mailer@handleQueuedMessage', compact('view', 'data', 'callback'), $queue);
    }
844424
protected function createPayload($job, $data = '', $queue = null)
    {
        if (is_object($job)) {
            $payload = json_encode([
                'job' => 'Globalis\PuppetSkilled\Queue\CallQueuedHandler@call',
                'data' => [
                    'commandName' => get_class($job),
                    'command' => serialize(clone $job),
                ],
            ]);
        } else {
            $payload = json_encode($this->createPlainPayload($job, $data));
        }

        if (JSON_ERROR_NONE !== json_last_error()) {
            throw new InvalidArgumentException('Unable to create payload: '.json_last_error_msg());
        }

        return $payload;
    }
111538
public function release($message, $delay = 0)
    {
        if (!$message instanceof QueueMessage) {
            $msg = new QueueMessage();
            $msg->setId($message);
        }

        $message->setDelay($delay);

        $this->driver->release($message);
    }
1216170
protected function pushToDatabase($queue, $payload, $delay = 0, $attempts = 0)
    {
        return $this->db->name($this->table)->insertGetId([
            'queue'        => $this->getQueue($queue),
            'attempts'     => $attempts,
            'reserved_at'  => null,
            'available_at' => $this->availableAt($delay),
            'created_at'   => $this->currentTime(),
            'payload'      => $payload,
        ]);
    }
596866
public function later($delay, $job, $data = '', $queue = null)
    {
        $delay = $this->getSeconds($delay);

        $payload = $this->createPayload($job, $data, $queue);

        return $this->pushRaw($payload, $queue, compact('delay'));
    }
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)
        );
    }
857596
public static function later($delay, $job, $data = '', $queue = null, $connection = null)
    {
        return static::$instance->connection($connection)->later($delay, $job, $data, $queue);
    }
137612
public function push($job, $data = '', $queue = null)
    {
        $this->lastPushed = $job;

        return parent::push($job, $data, $queue);
    }
129859
public function release(
        Job $job,
        $priority = null,
        $delay = null
    )
    {
        $priority = $priority ?? PheanstalkInterface::DEFAULT_PRIORITY;
        $delay = $delay ?? PheanstalkInterface::DEFAULT_DELAY;

        return $this->client->release($job, $priority, $delay);
    }
908314
public function publish($queueName, $queueData )
    {
        $time = time();
        $data = [
            'queue_name' => $queueData['queue_name'],
            'worker' => $queueData['worker'],
            'params' => $queueData['params'],
            'create_time' => $time,
        ];
        return $this->pushRaw($this->createPayload($data), $queueName);
    }
989905
protected function pushToDatabase($queue, $payload, $delay = 0, $attempts = 0)
    {
        return $this->database->table($this->table)->insertGetId($this->buildDatabaseRecord(
            $this->getQueue($queue), $payload, $this->availableAt($delay), $attempts
        ));
    }
35289
public function execute($data)
    {
        $client = $this->pool_chan->pop();
        try {
            $arguments = $data['arguments'];
            $data['result'] = call_user_func_array([$client, $data['name']], $arguments);
        } catch (\RedisException $e) {
            $this->reconnect($client);
            $this->execute($data);
            // $this->commands->push($data);
        }
        //回归连接
        $this->pushToPool($client);
        return $data['result'];
    }