AI 助理
备案控制台
开发者社区 云原生 云消息队列 文章 正文

laravel的消息队列剖析

2018-01-07 2150
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介：

laravel的消息队列剖析

这篇来自于看到朋友转的58沈剑的一篇文章：1分钟实现“延迟消息”功能

在实际工作中也不止遇见过一次这个问题，我在想着以前是怎么处理的呢？我记得当初在上家公司的时候直接使用的是laravel的queue来实现的。当然，这里说的laravel的queue实际上也是基于redis的队列实现的。正好今天遇上这个问题，追下底层机制。

使用如下：http://d.laravel-china.org/docs/5.3/queues

// 创建10分钟后执行的任务

$job = (new ProcessPodcast($pocast))
            ->delay(Carbon::now()->addMinutes(10));

dispatch($job);
//启动队列命令
php artisan queue:work

分发部分

首先看dispatch这边做的事情：

dispatch函数首先就是调用

return app(Dispatcher::class)->dispatch($job);
// Illuminate\Contracts\Bus\Dispatcher

首先需要理解这里的Dispatcher::class 实际注入的是哪个类。

看到vendor/laravel/framework/src/Illuminate/Bus/BusServiceProvider.php:26，有

public function register()
{
    $this->app->singleton('Illuminate\Bus\Dispatcher', function ($app) {
        return new Dispatcher($app, function ($connection = null) use ($app) {
            return $app['Illuminate\Contracts\Queue\Factory']->connection($connection);
        });
    });

    $this->app->alias(
        'Illuminate\Bus\Dispatcher', 'Illuminate\Contracts\Bus\Dispatcher'
    );

    $this->app->alias(
        'Illuminate\Bus\Dispatcher', 'Illuminate\Contracts\Bus\QueueingDispatcher'
    );
}

所以最后是实例化了Illuminate\Bus\Dispatcher

看看它的dispatch函数做了啥？

public function dispatch($command)
{
    if ($this->queueResolver && $this->commandShouldBeQueued($command)) {
        return $this->dispatchToQueue($command);
    } else {
        return $this->dispatchNow($command);
    }
}

假设我们的dispatch是基于队列的（ShouldQueue）。那么就是走dispatchToQueue，最终，走的是pushCommandToQueue

protected function pushCommandToQueue($queue, $command)
{
    ...

    if (isset($command->delay)) {
        return $queue->later($command->delay, $command);
    }
    ...
}

这里的queue就是队列的范畴了，假设我们用的队列是redis。（队列的解析器就是singleton的时候传入的Cluster）。最终这里落入的是vendor/laravel/framework/src/Illuminate/Queue/RedisQueue.php:111的

public function later($delay, $job, $data = '', $queue = null)
{
    $payload = $this->createPayload($job, $data);

    $this->getConnection()->zadd(
        $this->getQueue($queue).':delayed', $this->getTime() + $this->getSeconds($delay), $payload
    );

    return Arr::get(json_decode($payload, true), 'id');
}

这下就看清楚了：

laravel的延迟队列，使用的是zadd命令，往{$queue}:delayed，中插入一条job信息，它的score是执行时间。

（得到这条结论还真tmd是不容易）

队列监听部分

队列监听命令来自于： php artisan queue:work

命令行的入口就不追踪了，直接到vendor/laravel/framework/src/Illuminate/Queue/Console/WorkCommand.php:29 类

protected function runWorker($connection, $queue)
{
    $this->worker->setCache($this->laravel['cache']->driver());

    return $this->worker->{$this->option('once') ? 'runNextJob' : 'daemon'}(
        $connection, $queue, $this->gatherWorkerOptions()
    );
}

这里的daemon和runNextJob是只跑一次还是持续跑的意思，我们当然假定是以daemon的形式在跑。

这里的worker是vendor/laravel/framework/src/Illuminate/Queue/Worker.php:78

public function daemon($connectionName, $queue, WorkerOptions $options)
{
    $lastRestart = $this->getTimestampOfLastQueueRestart();

    while (true) {
        $this->registerTimeoutHandler($options);

        if ($this->daemonShouldRun($options)) {
            $this->runNextJob($connectionName, $queue, $options);
        } else {
            $this->sleep($options->sleep);
        }

        if ($this->memoryExceeded($options->memory) ||
            $this->queueShouldRestart($lastRestart)) {
            $this->stop();
        }
    }
}

这里的代码就值得我们自己写deamon的时候来参考了，它考虑了timeout，考虑了memory的情况。

而runNextJob的命令实际上就很清晰了

public function runNextJob($connectionName, $queue, WorkerOptions $options)
{
    ...
        $job = $this->getNextJob(
            $this->manager->connection($connectionName), $queue
        );

        ...
            return $this->process(
                $connectionName, $job, $options
            );
    ...
}

这里的Manager对应的是QueueManager, 这个类内部会创建一个connector（vendor/laravel/framework/src/Illuminate/Queue/Connectors/RedisConnector.php:30）

public function connect(array $config)
{
    return new RedisQueue(
        $this->redis, $config['queue'],
        Arr::get($config, 'connection', $this->connection),
        Arr::get($config, 'retry_after', 60)
    );
}

看到这里就明白了，最后还是掉落到RedisQueue中。 很好，和我们前面的任务分发终于对上了，圈子差不多画完了，我们可以看到曙光了。

追到RedisQueue里面，看它的pop行为。

public function pop($queue = null)
{
    $original = $queue ?: $this->default;

    $queue = $this->getQueue($queue);

    $this->migrateExpiredJobs($queue.':delayed', $queue);

    if (! is_null($this->expire)) {
        $this->migrateExpiredJobs($queue.':reserved', $queue);
    }

    list($job, $reserved) = $this->getConnection()->eval(
        LuaScripts::pop(), 2, $queue, $queue.':reserved', $this->getTime() + $this->expire
    );

    if ($reserved) {
        return new RedisJob($this->container, $this, $job, $reserved, $original);
    }
}

这段就是精华了。它做了什么事情呢？

先看migrateExpiredJobs:

public function migrateExpiredJobs($from, $to)
{
    $this->getConnection()->eval(
        LuaScripts::migrateExpiredJobs(), 2, $from, $to, $this->getTime()
    );
}

这里的eval就是对应redis的eval操作，https://redis.io/commands/eval，2是说明后面有两个key，最后一个getTime()获取的是arg。
下面就看lua脚本了。

public static function migrateExpiredJobs()
{
    return <<<'LUA'
local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])
if(next(val) ~= nil) then
redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)
for i = 1, #val, 100 do
    redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
end
end
return true
LUA;
}

结合起来看就是：

  • 使用zrangebyscore 和zremrangebyrank 从{queue}:delayed 队列中，从-inf到now的任务拿出来。
  • 用rpush的方式存入到默认queue中（后续就是放入到{queue}:reserved ）

这个zrangebyscore就是判断延迟任务是否应该执行的操作了。

然后就进行的是

list($job, $reserved) = $this->getConnection()->eval(
    LuaScripts::pop(), 2, $queue, $queue.':reserved', $this->getTime() + $this->expire
);

这里的LuaScripts::pop()如下：

public static function pop()
{
    return <<<'LUA'
local job = redis.call('lpop', KEYS[1])
local reserved = false
if(job ~= false) then
reserved = cjson.decode(job)
reserved['attempts'] = reserved['attempts'] + 1
reserved = cjson.encode(reserved)
redis.call('zadd', KEYS[2], ARGV[1], reserved)
end
return {job, reserved}
LUA;
}

做了下面操作：

  • 把默认队列中的任务lpop出来
  • 将他的attempts次数+1
  • zadd 存入{queue}:reserved 队列，score为now+60(默认的过期时间)

最后，我就返回这个job，这里结束了getNextJob的过程

process过程就是调用了一下：vendor/laravel/framework/src/Illuminate/Queue/Worker.php:187

public function process($connectionName, $job, WorkerOptions $options)
{
    try {
        $this->raiseBeforeJobEvent($connectionName, $job);

        $this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
            $connectionName, $job, (int) $options->maxTries
        );

        // Here we will fire off the job and let it process. We will catch any exceptions so
        // they can be reported to the developers logs, etc. Once the job is finished the
        // proper events will be fired to let any listeners know this job has finished.
        $job->fire();

        $this->raiseAfterJobEvent($connectionName, $job);
    } catch (Exception $e) {
        $this->handleJobException($connectionName, $job, $options, $e);
    } catch (Throwable $e) {
        $this->handleJobException(
            $connectionName, $job, $options, new FatalThrowableError($e)
        );
    }
}

$this->events->fire(new Events\JobProcessing(
    $connectionName, $job
));

这里的raiseBeforeJobEvent和raiseAfterJobEvent又是使用event和listener的形式来做处理的。这里的$this->events是vendor/laravel/framework/src/Illuminate/Events/Dispatcher.php:197

这里就是触发了一个Events\JobProcessing事件，我们现在要找到对应的lister：

答案是在QueueManager中定义的

/**
 * Register an event listener for the before job event.
 *
 * @param  mixed  $callback
 * @return void
 */
public function before($callback)
{
    $this->app['events']->listen(Events\JobProcessing::class, $callback);
}

/**
 * Register an event listener for the after job event.
 *
 * @param  mixed  $callback
 * @return void
 */
public function after($callback)
{
    $this->app['events']->listen(Events\JobProcessed::class, $callback);
}

换句话说，我们希望监听一个job开始和结束的时候，我们可以使用QueueManager的before，after来监听。比如发个邮件，唱唱小曲啥的。

那么这里我们，从{queue}:reserved中获取了job之后（这里的job是RedisJob），我们是什么时候触发的delete呢？是在

$job->fire();

这个fire是RedisJob(vendor/laravel/framework/src/Illuminate/Queue/Jobs/RedisJob.php)但继承来自vendor/laravel/framework/src/Illuminate/Queue/Jobs/Job.php:72， 经过调用CallQueuedHandler，最终会落到
vendor/laravel/framework/src/Illuminate/Queue/RedisQueue.php:154

public function deleteReserved($queue, $job)
{
    $this->getConnection()->zrem($this->getQueue($queue).':reserved', $job);
}

这里就是将job从{queue}:reserved 队列中删除。

至此，整个队列及延迟机制就处理完了。

实际

我们实际监听一下redis就可以验证结果：

// 使用dispatch
1489802272.491060 [0 127.0.0.1:63798] "SELECT" "0"
1489802272.491513 [0 127.0.0.1:63798] "ZADD" "queues:default:delayed" "1489802332" "{\"job\":\"Illuminate\\\\Queue\\\\CallQueuedHandler@call\",\"data\":{\"commandName\":\"App\\\\Jobs\\\\DelayTestJob\",\"command\":\"O:21:\\\"App\\\\Jobs\\\\DelayTestJob\\\":4:{s:6:\\\"\\u0000*\\u0000job\\\";N;s:10:\\\"connection\\\";N;s:5:\\\"queue\\\";N;s:5:\\\"delay\\\";O:13:\\\"Carbon\\\\Carbon\\\":3:{s:4:\\\"date\\\";s:26:\\\"2017-03-18 01:58:52.000000\\\";s:13:\\\"timezone_type\\\";i:3;s:8:\\\"timezone\\\";s:3:\\\"UTC\\\";}}\"},\"id\":\"q7ss6fRgCbMNHhCv6gOXX0Or7B43blU9\",\"attempts\":1}"


// 1分钟后
1489802333.957500 [0 127.0.0.1:63792] "EVAL" "local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])\nif(next(val) ~= nil) then\n    redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)\n    for i = 1, #val, 100 do\n        redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))\n    end\nend\nreturn true" "2" "queues:default:delayed" "queues:default" "1489802333"
1489802333.957563 [0 lua] "zrangebyscore" "queues:default:delayed" "-inf" "1489802333"
1489802333.957586 [0 lua] "zremrangebyrank" "queues:default:delayed" "0" "0"
1489802333.958628 [0 lua] "rpush" "queues:default" "{\"job\":\"Illuminate\\\\Queue\\\\CallQueuedHandler@call\",\"data\":{\"commandName\":\"App\\\\Jobs\\\\DelayTestJob\",\"command\":\"O:21:\\\"App\\\\Jobs\\\\DelayTestJob\\\":4:{s:6:\\\"\\u0000*\\u0000job\\\";N;s:10:\\\"connection\\\";N;s:5:\\\"queue\\\";N;s:5:\\\"delay\\\";O:13:\\\"Carbon\\\\Carbon\\\":3:{s:4:\\\"date\\\";s:26:\\\"2017-03-18 01:58:52.000000\\\";s:13:\\\"timezone_type\\\";i:3;s:8:\\\"timezone\\\";s:3:\\\"UTC\\\";}}\"},\"id\":\"q7ss6fRgCbMNHhCv6gOXX0Or7B43blU9\",\"attempts\":1}"
1489802333.959572 [0 127.0.0.1:63792] "EVAL" "local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])\nif(next(val) ~= nil) then\n    redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)\n    for i = 1, #val, 100 do\n        redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))\n    end\nend\nreturn true" "2" "queues:default:reserved" "queues:default" "1489802333"
1489802333.959672 [0 lua] "zrangebyscore" "queues:default:reserved" "-inf" "1489802333"
1489802333.959866 [0 127.0.0.1:63792] "EVAL" "local job = redis.call('lpop', KEYS[1])\nlocal reserved = false\nif(job ~= false) then\n    reserved = cjson.decode(job)\n    reserved['attempts'] = reserved['attempts'] + 1\n    reserved = cjson.encode(reserved)\n    redis.call('zadd', KEYS[2], ARGV[1], reserved)\nend\nreturn {job, reserved}" "2" "queues:default" "queues:default:reserved" "1489802343"
1489802333.959938 [0 lua] "lpop" "queues:default"
1489802333.959965 [0 lua] "zadd" "queues:default:reserved" "1489802343" "{\"id\":\"q7ss6fRgCbMNHhCv6gOXX0Or7B43blU9\",\"attempts\":2,\"data\":{\"command\":\"O:21:\\\"App\\\\Jobs\\\\DelayTestJob\\\":4:{s:6:\\\"\\u0000*\\u0000job\\\";N;s:10:\\\"connection\\\";N;s:5:\\\"queue\\\";N;s:5:\\\"delay\\\";O:13:\\\"Carbon\\\\Carbon\\\":3:{s:4:\\\"date\\\";s:26:\\\"2017-03-18 01:58:52.000000\\\";s:13:\\\"timezone_type\\\";i:3;s:8:\\\"timezone\\\";s:3:\\\"UTC\\\";}}\",\"commandName\":\"App\\\\Jobs\\\\DelayTestJob\"},\"job\":\"Illuminate\\\\Queue\\\\CallQueuedHandler@call\"}"
1489802333.963223 [0 127.0.0.1:63792] "ZREM" "queues:default:reserved" "{\"id\":\"q7ss6fRgCbMNHhCv6gOXX0Or7B43blU9\",\"attempts\":2,\"data\":{\"command\":\"O:21:\\\"App\\\\Jobs\\\\DelayTestJob\\\":4:{s:6:\\\"\\u0000*\\u0000job\\\";N;s:10:\\\"connection\\\";N;s:5:\\\"queue\\\";N;s:5:\\\"delay\\\";O:13:\\\"Carbon\\\\Carbon\\\":3:{s:4:\\\"date\\\";s:26:\\\"2017-03-18 01:58:52.000000\\\";s:13:\\\"timezone_type\\\";i:3;s:8:\\\"timezone\\\";s:3:\\\"UTC\\\";}}\",\"commandName\":\"App\\\\Jobs\\\\DelayTestJob\"},\"job\":\"Illuminate\\\\Queue\\\\CallQueuedHandler@call\"}"

精简下路径就是：

// 第一步：先往delayed队列中插入job
1489802272.491513 [0 127.0.0.1:63798] "ZADD" "queues:default:delayed" "1489802332" {job}

// 第二步，将delayed队列中到期的job取出，并且rpush进default队列
1489802333.957563 [0 lua] "zrangebyscore" "queues:default:delayed" "-inf" "1489802333"
1489802333.957586 [0 lua] "zremrangebyrank" "queues:default:delayed" "0" "0"
1489802333.958628 [0 lua] "rpush" "queues:default" {job}

// 第三步，从default队列中lpop出job
1489802333.959938 [0 lua] "lpop" "queues:default"

// 第四步，zadd到default:reserved
1489802333.959965 [0 lua] "zadd" "queues:default:reserved" "1489802343" {job}

// 第五步，程序处理这个job

// 第六步，讲job从default:reserved中删除
1489802333.963223 [0 127.0.0.1:63792] "ZREM" "queues:default:reserved" {job}

符合预期。

总结

laravel这边的延迟队列使用了三个队列。

  • queue:default:delayed // 存储延迟任务
  • queue:default // 存储“生”任务，就是未处理任务
  • queue:default:reserved // 存储待处理任务

任务在三个队列中进行轮转，最后一定进入到queue:default:reserved，并且成功后把任务从这个队列中删除。

其间还使用了lua脚本，所以至少laravel5.3（本文的laravel环境）在无lua脚本支持的redis版本是跑不了的。

它用三个队列把所有的步骤给原子了，所以并没有使用multi等操作。也是防止了锁的使用把。每一步操作失败了，都会有后续的步骤继续帮忙完成，记录等行为的。


本文转自轩脉刃博客园博客，原文链接：http://www.cnblogs.com/yjf512/p/6571941.html，如需转载请自行联系原作者

文章标签：
云数据库 Tair（兼容 Redis）
PHP
消息中间件
NoSQL
Redis
存储
容器
相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务，基于高可靠双机热备架构及可无缝扩展的集群架构，满足高读写性能场景及容量需弹性变配的业务需求。 产品详情：https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验：数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引，您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接，领取免费ECS&amp;RDS资源，30分钟完成数据库上云实战！https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
技术mix呢
目录
相关文章
DreamSpark
|
2月前
|
消息中间件 Serverless 数据安全/隐私保护
开发者如何使用云消息队列 RabbitMQ 版
【10月更文挑战第13天】开发者如何使用云消息队列 RabbitMQ 版
DreamSpark
111 10
DreamSpark
|
2月前
|
消息中间件 监控 Java
开发者如何使用云消息队列 RocketMQ 版
【10月更文挑战第12天】开发者如何使用云消息队列 RocketMQ 版
DreamSpark
222 4
yuanzhengme
|
2月前
|
消息中间件 中间件 UED
为什么需要消息队列中间件？
为什么需要消息队列中间件？
yuanzhengme
55 4
疯狂的猿
|
7月前
|
消息中间件 NoSQL 中间件
常用的消息队列和中间件都有哪些
常用的消息队列和中间件都有哪些
疯狂的猿
320 7
不摸鱼的程序员
|
7月前
|
消息中间件 存储 网络协议
手写消息队列（基于RabbitMQ）
手写消息队列（基于RabbitMQ）
不摸鱼的程序员
86 0
极客李华
|
7月前
|
消息中间件 存储 Kafka
MQ消息队列学习入门
MQ消息队列学习入门
极客李华
123 0
憧憬blog
|
7月前
|
消息中间件 数据可视化 Ubuntu
php laravel5.5使用rabbitmq消息队列
php laravel5.5使用rabbitmq消息队列
憧憬blog
52 0
yangnk42
|
消息中间件 网络协议 Java
RabbitMQ消息队列的原理和实践
RabbitMQ消息队列的原理和实践
yangnk42
449 0
战斧.
|
消息中间件 自然语言处理 监控
消息队列选型——为什么选择RabbitMQ
消息队列选型——为什么选择RabbitMQ
战斧.
324 0
wu@55555
|
消息中间件 监控 大数据
RabbitMQ：什么是消息队列MQ？为什么使用消息队列MQ？入门MQ先学哪种？（一）
MQ(Message Queue)：消息队列，如今在各类业务场景中已经被广泛使用，特别在并发量日益增涨的业务和微服务架构中，消息队列能够帮助我们解决很多传统方式所不能解决的问题。 所以今天，我们就开始学习消息队列啦
wu@55555
447 0
RabbitMQ：什么是消息队列MQ？为什么使用消息队列MQ？入门MQ先学哪种？（一）
云原生

云消息队列

热门文章

最新文章

  • 1
    使用阿里云消息队列
  • 2
    ActiveMQ消息队列
  • 3
    Redis笔记（七）Java实现Redis消息队列
  • 4
    阿里云消息队列Kafka商业化：支持消息无缝迁移到云上
  • 5
    基于Redis实现分布式消息队列（3）
  • 6
    springboot整合mq接收消息队列
  • 7
    异步消息队列zeromq实现服务器间高性能通信
  • 8
    服务器设计笔记(3)-----消息队列
  • 9
    laravel的消息队列剖析
  • 10
    Linux进程间通信——使用消息队列
  • 1
    云消息队列 RabbitMQ 版实践解决方案评测
    89
  • 2
    RocketMQ体验测评
    73
  • 3
    关键成果与技术应用
    82
  • 4
    消息队列 MQ使用问题之如何使用DefaultMQPushConsumer来消费消息
    192
  • 5
    消息队列 MQ使用问题之遇到消费速度是固定的并且导致了堆积，该怎么办
    210
  • 6
    消息队列 MQ使用问题之如何在grpc客户端中设置负载均衡器
    62
  • 7
    消息队列 MQ使用问题之如何配置生产环境
    125
  • 8
    消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump，该怎么办
    133
  • 9
    消息队列 MQ使用问题之如何将RocketMQ中某个集群的topic迁移到另一个集群
    122
  • 10
    消息队列 MQ操作报错合集之建立连接时发生了超时错误，该如何解决
    631

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列RabbitMQ入门课程
  • 带您快速了解阿里云消息队列RocketMQ 5.0版
  • 消息队列 RocketMQ 全类型业务消息学习课程
  • 消息队列Kafka入门课程
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • 消息队列 Kafka 版差异化特性
  • 基于消息队列RocketMQ的大型分布式应用上云最佳实践
  • 云原生消息队列Apache RocketMQ

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速入门轻量消息队列（原MNS）
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
    • 下一篇
    DataWorks售前咨询