think-queue 解析上

    push($job, $data, $queue)
    Queue::push(Test::class, ['id' => 1], 'test');

[
    'job' => $job, // 要执行任务的类
    'data' => $data, // 任务数据
    'id'=>'xxxxx' //任务id
]

    later($delay, $job, $data, $queue)
    Queue::later(100, Test::class, ['id' => 1], 'test');

    //守护进程开启
    'worker_daemon_start' => [
        \app\index\behavior\WorkerDaemonStart::class
    ],
    //内存超出
    'worker_memory_exceeded' => [
        \app\index\behavior\WorkerMemoryExceeded::class
    ],
    //重启守护进程
    'worker_queue_restart' => [
        \app\index\behavior\WorkerQueueRestart::class
    ],
    //任务开始执行之前
    'worker_before_process' => [
        \app\index\behavior\WorkerBeforeProcess::class
    ],
    //任务延迟执行
    'worker_before_sleep' => [
        \app\index\behavior\WorkerBeforeSleep::class
    ],
    //任务执行失败
    'queue_failed' => [
        \app\index\behavior\QueueFailed::class
    ]

public function run(Output $output)
    {
        $output->write('<info>任务执行失败</info>', true);
    }

守护进程开启
任务延迟执行

<?php

class Index
{
    private $redis = null;

    public function __construct()
    {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
        $this->redis->select(10);
    }

    public function push($job, $data, $queue)
    {
        $payload = $this->createPayload($job, $data);
        $this->redis->rPush('queues:' . $queue, $payload);
    }

    public function later($delay, $job, $data, $queue)
    {
        $payload = $this->createPayload($job, $data);
        $this->redis->zAdd('queues:' . $queue . ':delayed', time() + $delay, $payload);
    }

    private function createPayload($job, $data)
    {
        $payload = $this->setMeta(json_encode(['job' => $job, 'data' => $data]), 'id', $this->random(32));
        return $this->setMeta($payload, 'attempts', 1);
    }

    private function setMeta($payload, $key, $value)
    {
        $payload = json_decode($payload, true);
        $payload[$key] = $value;
        $payload = json_encode($payload);

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

        return $payload;
    }

    private function random(int $length = 16): string
    {
        $str = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ';
        $randomString = '';
        for ($i = 0; $i < $length; $i++) {
            $randomString .= $str[rand(0, strlen($str) - 1)];
        }
        return $randomString;
    }
}

(new Index())->later(10, 'app\index\jobs\Test', ['id' => 1], 'test');

package main

import (
	"encoding/json"
	"github.com/garyburd/redigo/redis"
	"math/rand"
	"time"
)

type Payload struct {
	Id       string      `json:"id"`
	Job      string      `json:"job"`
	Data     interface{} `json:"data"`
	Attempts int         `json:"attempts"`
}

var RedisClient *redis.Pool

func init() {
	RedisClient = &redis.Pool{
		MaxIdle:     20,
		MaxActive:   500,
		IdleTimeout: time.Second * 100,
		Dial: func() (conn redis.Conn, e error) {
			c, err := redis.Dial("tcp", "127.0.0.1:6379")

			if err != nil {
				return nil, err
			}

			_, _ = c.Do("SELECT", 10)

			return c, nil
		},
	}

}

func main() {

	var data = make(map[string]interface{})
	data["id"] = "1"

	later(10, "app\\index\\jobs\\Test", data, "test")
}

func push(job string, data interface{}, queue string) {
	payload := createPayload(job, data)
	queueName := "queues:" + queue

	_, _ = RedisClient.Get().Do("rPush", queueName, payload)
}

func later(delay int, job string, data interface{}, queue string) {

	m, _ := time.ParseDuration("+1s")
	currentTime := time.Now()
	op := currentTime.Add(time.Duration(time.Duration(delay) * m)).Unix()
	createPayload(job, data)
	payload := createPayload(job, data)
	queueName := "queues:" + queue + ":delayed"

	_, _ = RedisClient.Get().Do("zAdd", queueName, op, payload)
}

// 创建指定格式的数据
func createPayload(job string, data interface{}) (payload string) {
	payload1 := &Payload{Job: job, Data: data, Id: random(32), Attempts: 1}

	jsonStr, _ := json.Marshal(payload1)

	return string(jsonStr)
}

// 创建随机字符串
func random(n int) string {

	var str = []rune("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")

	b := make([]rune, n)
	for i := range b {
		b[i] = str[rand.Intn(len(str))]
	}
	return string(b)
}