update
This commit is contained in:
yyh931018@qq.com
4
vendor/topthink/think-queue/.gitignore
vendored
Normal file
4
vendor/topthink/think-queue/.gitignore
vendored
Normal file
@@ -0,0 +1,4 @@
|
||||
/vendor/
|
||||
/.idea/
|
||||
/composer.lock
|
||||
/thinkphp/
|
||||
201
vendor/topthink/think-queue/LICENSE
vendored
Normal file
201
vendor/topthink/think-queue/LICENSE
vendored
Normal file
@@ -0,0 +1,201 @@
|
||||
Apache License
|
||||
Version 2.0, January 2004
|
||||
http://www.apache.org/licenses/
|
||||
|
||||
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
|
||||
|
||||
1. Definitions.
|
||||
|
||||
"License" shall mean the terms and conditions for use, reproduction,
|
||||
and distribution as defined by Sections 1 through 9 of this document.
|
||||
|
||||
"Licensor" shall mean the copyright owner or entity authorized by
|
||||
the copyright owner that is granting the License.
|
||||
|
||||
"Legal Entity" shall mean the union of the acting entity and all
|
||||
other entities that control, are controlled by, or are under common
|
||||
control with that entity. For the purposes of this definition,
|
||||
"control" means (i) the power, direct or indirect, to cause the
|
||||
direction or management of such entity, whether by contract or
|
||||
otherwise, or (ii) ownership of fifty percent (50%) or more of the
|
||||
outstanding shares, or (iii) beneficial ownership of such entity.
|
||||
|
||||
"You" (or "Your") shall mean an individual or Legal Entity
|
||||
exercising permissions granted by this License.
|
||||
|
||||
"Source" form shall mean the preferred form for making modifications,
|
||||
including but not limited to software source code, documentation
|
||||
source, and configuration files.
|
||||
|
||||
"Object" form shall mean any form resulting from mechanical
|
||||
transformation or translation of a Source form, including but
|
||||
not limited to compiled object code, generated documentation,
|
||||
and conversions to other media types.
|
||||
|
||||
"Work" shall mean the work of authorship, whether in Source or
|
||||
Object form, made available under the License, as indicated by a
|
||||
copyright notice that is included in or attached to the work
|
||||
(an example is provided in the Appendix below).
|
||||
|
||||
"Derivative Works" shall mean any work, whether in Source or Object
|
||||
form, that is based on (or derived from) the Work and for which the
|
||||
editorial revisions, annotations, elaborations, or other modifications
|
||||
represent, as a whole, an original work of authorship. For the purposes
|
||||
of this License, Derivative Works shall not include works that remain
|
||||
separable from, or merely link (or bind by name) to the interfaces of,
|
||||
the Work and Derivative Works thereof.
|
||||
|
||||
"Contribution" shall mean any work of authorship, including
|
||||
the original version of the Work and any modifications or additions
|
||||
to that Work or Derivative Works thereof, that is intentionally
|
||||
submitted to Licensor for inclusion in the Work by the copyright owner
|
||||
or by an individual or Legal Entity authorized to submit on behalf of
|
||||
the copyright owner. For the purposes of this definition, "submitted"
|
||||
means any form of electronic, verbal, or written communication sent
|
||||
to the Licensor or its representatives, including but not limited to
|
||||
communication on electronic mailing lists, source code control systems,
|
||||
and issue tracking systems that are managed by, or on behalf of, the
|
||||
Licensor for the purpose of discussing and improving the Work, but
|
||||
excluding communication that is conspicuously marked or otherwise
|
||||
designated in writing by the copyright owner as "Not a Contribution."
|
||||
|
||||
"Contributor" shall mean Licensor and any individual or Legal Entity
|
||||
on behalf of whom a Contribution has been received by Licensor and
|
||||
subsequently incorporated within the Work.
|
||||
|
||||
2. Grant of Copyright License. Subject to the terms and conditions of
|
||||
this License, each Contributor hereby grants to You a perpetual,
|
||||
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
|
||||
copyright license to reproduce, prepare Derivative Works of,
|
||||
publicly display, publicly perform, sublicense, and distribute the
|
||||
Work and such Derivative Works in Source or Object form.
|
||||
|
||||
3. Grant of Patent License. Subject to the terms and conditions of
|
||||
this License, each Contributor hereby grants to You a perpetual,
|
||||
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
|
||||
(except as stated in this section) patent license to make, have made,
|
||||
use, offer to sell, sell, import, and otherwise transfer the Work,
|
||||
where such license applies only to those patent claims licensable
|
||||
by such Contributor that are necessarily infringed by their
|
||||
Contribution(s) alone or by combination of their Contribution(s)
|
||||
with the Work to which such Contribution(s) was submitted. If You
|
||||
institute patent litigation against any entity (including a
|
||||
cross-claim or counterclaim in a lawsuit) alleging that the Work
|
||||
or a Contribution incorporated within the Work constitutes direct
|
||||
or contributory patent infringement, then any patent licenses
|
||||
granted to You under this License for that Work shall terminate
|
||||
as of the date such litigation is filed.
|
||||
|
||||
4. Redistribution. You may reproduce and distribute copies of the
|
||||
Work or Derivative Works thereof in any medium, with or without
|
||||
modifications, and in Source or Object form, provided that You
|
||||
meet the following conditions:
|
||||
|
||||
(a) You must give any other recipients of the Work or
|
||||
Derivative Works a copy of this License; and
|
||||
|
||||
(b) You must cause any modified files to carry prominent notices
|
||||
stating that You changed the files; and
|
||||
|
||||
(c) You must retain, in the Source form of any Derivative Works
|
||||
that You distribute, all copyright, patent, trademark, and
|
||||
attribution notices from the Source form of the Work,
|
||||
excluding those notices that do not pertain to any part of
|
||||
the Derivative Works; and
|
||||
|
||||
(d) If the Work includes a "NOTICE" text file as part of its
|
||||
distribution, then any Derivative Works that You distribute must
|
||||
include a readable copy of the attribution notices contained
|
||||
within such NOTICE file, excluding those notices that do not
|
||||
pertain to any part of the Derivative Works, in at least one
|
||||
of the following places: within a NOTICE text file distributed
|
||||
as part of the Derivative Works; within the Source form or
|
||||
documentation, if provided along with the Derivative Works; or,
|
||||
within a display generated by the Derivative Works, if and
|
||||
wherever such third-party notices normally appear. The contents
|
||||
of the NOTICE file are for informational purposes only and
|
||||
do not modify the License. You may add Your own attribution
|
||||
notices within Derivative Works that You distribute, alongside
|
||||
or as an addendum to the NOTICE text from the Work, provided
|
||||
that such additional attribution notices cannot be construed
|
||||
as modifying the License.
|
||||
|
||||
You may add Your own copyright statement to Your modifications and
|
||||
may provide additional or different license terms and conditions
|
||||
for use, reproduction, or distribution of Your modifications, or
|
||||
for any such Derivative Works as a whole, provided Your use,
|
||||
reproduction, and distribution of the Work otherwise complies with
|
||||
the conditions stated in this License.
|
||||
|
||||
5. Submission of Contributions. Unless You explicitly state otherwise,
|
||||
any Contribution intentionally submitted for inclusion in the Work
|
||||
by You to the Licensor shall be under the terms and conditions of
|
||||
this License, without any additional terms or conditions.
|
||||
Notwithstanding the above, nothing herein shall supersede or modify
|
||||
the terms of any separate license agreement you may have executed
|
||||
with Licensor regarding such Contributions.
|
||||
|
||||
6. Trademarks. This License does not grant permission to use the trade
|
||||
names, trademarks, service marks, or product names of the Licensor,
|
||||
except as required for reasonable and customary use in describing the
|
||||
origin of the Work and reproducing the content of the NOTICE file.
|
||||
|
||||
7. Disclaimer of Warranty. Unless required by applicable law or
|
||||
agreed to in writing, Licensor provides the Work (and each
|
||||
Contributor provides its Contributions) on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
implied, including, without limitation, any warranties or conditions
|
||||
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
|
||||
PARTICULAR PURPOSE. You are solely responsible for determining the
|
||||
appropriateness of using or redistributing the Work and assume any
|
||||
risks associated with Your exercise of permissions under this License.
|
||||
|
||||
8. Limitation of Liability. In no event and under no legal theory,
|
||||
whether in tort (including negligence), contract, or otherwise,
|
||||
unless required by applicable law (such as deliberate and grossly
|
||||
negligent acts) or agreed to in writing, shall any Contributor be
|
||||
liable to You for damages, including any direct, indirect, special,
|
||||
incidental, or consequential damages of any character arising as a
|
||||
result of this License or out of the use or inability to use the
|
||||
Work (including but not limited to damages for loss of goodwill,
|
||||
work stoppage, computer failure or malfunction, or any and all
|
||||
other commercial damages or losses), even if such Contributor
|
||||
has been advised of the possibility of such damages.
|
||||
|
||||
9. Accepting Warranty or Additional Liability. While redistributing
|
||||
the Work or Derivative Works thereof, You may choose to offer,
|
||||
and charge a fee for, acceptance of support, warranty, indemnity,
|
||||
or other liability obligations and/or rights consistent with this
|
||||
License. However, in accepting such obligations, You may act only
|
||||
on Your own behalf and on Your sole responsibility, not on behalf
|
||||
of any other Contributor, and only if You agree to indemnify,
|
||||
defend, and hold each Contributor harmless for any liability
|
||||
incurred by, or claims asserted against, such Contributor by reason
|
||||
of your accepting any such warranty or additional liability.
|
||||
|
||||
END OF TERMS AND CONDITIONS
|
||||
|
||||
APPENDIX: How to apply the Apache License to your work.
|
||||
|
||||
To apply the Apache License to your work, attach the following
|
||||
boilerplate notice, with the fields enclosed by brackets "{}"
|
||||
replaced with your own identifying information. (Don't include
|
||||
the brackets!) The text should be enclosed in the appropriate
|
||||
comment syntax for the file format. We also recommend that a
|
||||
file or class name and description of purpose be included on the
|
||||
same "printed page" as the copyright notice for easier
|
||||
identification within third-party archives.
|
||||
|
||||
Copyright {yyyy} {name of copyright owner}
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
132
vendor/topthink/think-queue/README.md
vendored
Normal file
132
vendor/topthink/think-queue/README.md
vendored
Normal file
@@ -0,0 +1,132 @@
|
||||
# think-queue
|
||||
|
||||
## 安装
|
||||
> composer require topthink/think-queue
|
||||
|
||||
## 配置
|
||||
> 配置文件位于 `application/extra/queue.php`
|
||||
### 公共配置
|
||||
|
||||
```
|
||||
[
|
||||
'connector'=>'sync' //驱动类型，可选择 sync(默认):同步执行，database:数据库驱动,redis:Redis驱动,topthink:Topthink驱动
|
||||
//或其他自定义的完整的类名
|
||||
]
|
||||
```
|
||||
|
||||
### 驱动配置
|
||||
> 各个驱动的具体可用配置项在`think\queue\connector`目录下各个驱动类里的`options`属性中，写在上面的`queue`配置里即可覆盖
|
||||
|
||||
|
||||
## 使用 Database
|
||||
> 创建如下数据表
|
||||
|
||||
```
|
||||
CREATE TABLE `prefix_jobs` (
|
||||
`id` int(11) NOT NULL AUTO_INCREMENT,
|
||||
`queue` varchar(255) NOT NULL,
|
||||
`payload` longtext NOT NULL,
|
||||
`attempts` tinyint(3) unsigned NOT NULL,
|
||||
`reserved` tinyint(3) unsigned NOT NULL,
|
||||
`reserved_at` int(10) unsigned DEFAULT NULL,
|
||||
`available_at` int(10) unsigned NOT NULL,
|
||||
`created_at` int(10) unsigned NOT NULL,
|
||||
PRIMARY KEY (`id`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
|
||||
```
|
||||
|
||||
## 创建任务类
|
||||
> 单模块项目推荐使用 `app\job` 作为任务类的命名空间
|
||||
> 多模块项目可用使用 `app\module\job` 作为任务类的命名空间
|
||||
> 也可以放在任意可以自动加载到的地方
|
||||
|
||||
任务类不需继承任何类，如果这个类只有一个任务，那么就只需要提供一个`fire`方法就可以了，如果有多个小任务，就写多个方法，下面发布任务的时候会有区别
|
||||
每个方法会传入两个参数 `think\queue\Job $job`（当前的任务对象） 和 `$data`（发布任务时自定义的数据）
|
||||
|
||||
还有个可选的任务失败执行的方法 `failed` 传入的参数为`$data`（发布任务时自定义的数据）
|
||||
|
||||
### 下面写两个例子
|
||||
|
||||
```
|
||||
namespace app\job;
|
||||
|
||||
use think\queue\Job;
|
||||
|
||||
class Job1{
|
||||
|
||||
public function fire(Job $job, $data){
|
||||
|
||||
//....这里执行具体的任务
|
||||
|
||||
if ($job->attempts() > 3) {
|
||||
//通过这个方法可以检查这个任务已经重试了几次了
|
||||
}
|
||||
|
||||
|
||||
//如果任务执行成功后 记得删除任务，不然这个任务会重复执行，直到达到最大重试次数后失败后，执行failed方法
|
||||
$job->delete();
|
||||
|
||||
// 也可以重新发布这个任务
|
||||
$job->release($delay); //$delay为延迟时间
|
||||
|
||||
}
|
||||
|
||||
public function failed($data){
|
||||
|
||||
// ...任务达到最大重试次数后，失败了
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
```
|
||||
|
||||
```
|
||||
|
||||
namespace app\lib\job;
|
||||
|
||||
use think\queue\Job;
|
||||
|
||||
class Job2{
|
||||
|
||||
public function task1(Job $job, $data){
|
||||
|
||||
|
||||
}
|
||||
|
||||
public function task2(Job $job, $data){
|
||||
|
||||
|
||||
}
|
||||
|
||||
public function failed($data){
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
```
|
||||
|
||||
|
||||
## 发布任务
|
||||
> `think\Queue::push($job, $data = '', $queue = null)` 和 `think\Queue::later($delay, $job, $data = '', $queue = null)` 两个方法，前者是立即执行，后者是在`$delay`秒后执行
|
||||
|
||||
`$job` 是任务名
|
||||
单模块的，且命名空间是`app\job`的，比如上面的例子一,写`Job1`类名即可
|
||||
多模块的，且命名空间是`app\module\job`的，写`model/Job1`即可
|
||||
其他的需要些完整的类名，比如上面的例子二，需要写完整的类名`app\lib\job\Job2`
|
||||
如果一个任务类里有多个小任务的话，如上面的例子二，需要用@+方法名`app\lib\job\Job2@task1`、`app\lib\job\Job2@task2`
|
||||
|
||||
`$data` 是你要传到任务里的参数
|
||||
|
||||
`$queue` 队列名，指定这个任务是在哪个队列上执行，同下面监控队列的时候指定的队列名,可不填
|
||||
|
||||
## 监听任务并执行
|
||||
|
||||
> php think queue:listen
|
||||
|
||||
> php think queue:work --daemon（不加--daemon为执行单个任务）
|
||||
|
||||
两种，具体的可选参数可以输入命令加 --help 查看
|
||||
|
||||
>可配合supervisor使用，保证进程常驻
|
||||
32
vendor/topthink/think-queue/composer.json
vendored
Normal file
32
vendor/topthink/think-queue/composer.json
vendored
Normal file
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"name": "topthink/think-queue",
|
||||
"description": "The ThinkPHP5 Queue Package",
|
||||
"type": "think-extend",
|
||||
"authors": [
|
||||
{
|
||||
"name": "yunwuxin",
|
||||
"email": "448901948@qq.com"
|
||||
}
|
||||
],
|
||||
"license": "Apache-2.0",
|
||||
"autoload": {
|
||||
"psr-4": {
|
||||
"think\\": "src"
|
||||
},
|
||||
"files": [
|
||||
"src/common.php"
|
||||
]
|
||||
},
|
||||
"require": {
|
||||
"topthink/think-helper": ">=1.0.4",
|
||||
"topthink/think-installer": ">=1.0.10"
|
||||
},
|
||||
"require-dev": {
|
||||
"topthink/framework": "~5.0.0"
|
||||
},
|
||||
"extra": {
|
||||
"think-config": {
|
||||
"queue": "src/config.php"
|
||||
}
|
||||
}
|
||||
}
|
||||
49
vendor/topthink/think-queue/src/Queue.php
vendored
Normal file
49
vendor/topthink/think-queue/src/Queue.php
vendored
Normal file
@@ -0,0 +1,49 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think;
|
||||
|
||||
use think\helper\Str;
|
||||
use think\queue\Connector;
|
||||
|
||||
/**
|
||||
* Class Queue
|
||||
* @package think\queue
|
||||
*
|
||||
* @method static push($job, $data = '', $queue = null)
|
||||
* @method static later($delay, $job, $data = '', $queue = null)
|
||||
* @method static pop($queue = null)
|
||||
* @method static marshal()
|
||||
*/
|
||||
class Queue
|
||||
{
|
||||
/** @var Connector */
|
||||
protected static $connector;
|
||||
|
||||
private static function buildConnector()
|
||||
{
|
||||
$options = Config::get('queue');
|
||||
$type = !empty($options['connector']) ? $options['connector'] : 'Sync';
|
||||
|
||||
if (!isset(self::$connector)) {
|
||||
|
||||
$class = false !== strpos($type, '\\') ? $type : '\\think\\queue\\connector\\' . Str::studly($type);
|
||||
|
||||
self::$connector = new $class($options);
|
||||
}
|
||||
return self::$connector;
|
||||
}
|
||||
|
||||
public static function __callStatic($name, $arguments)
|
||||
{
|
||||
return call_user_func_array([self::buildConnector(), $name], $arguments);
|
||||
}
|
||||
}
|
||||
36
vendor/topthink/think-queue/src/common.php
vendored
Normal file
36
vendor/topthink/think-queue/src/common.php
vendored
Normal file
@@ -0,0 +1,36 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
\think\Console::addDefaultCommands([
|
||||
"think\\queue\\command\\Work",
|
||||
"think\\queue\\command\\Restart",
|
||||
"think\\queue\\command\\Listen",
|
||||
"think\\queue\\command\\Subscribe"
|
||||
]);
|
||||
|
||||
if (!function_exists('queue')) {
|
||||
|
||||
/**
|
||||
* 添加到队列
|
||||
* @param $job
|
||||
* @param string $data
|
||||
* @param int $delay
|
||||
* @param null $queue
|
||||
*/
|
||||
function queue($job, $data = '', $delay = 0, $queue = null)
|
||||
{
|
||||
if ($delay > 0) {
|
||||
\think\Queue::later($delay, $job, $data, $queue);
|
||||
} else {
|
||||
\think\Queue::push($job, $data, $queue);
|
||||
}
|
||||
}
|
||||
}
|
||||
14
vendor/topthink/think-queue/src/config.php
vendored
Normal file
14
vendor/topthink/think-queue/src/config.php
vendored
Normal file
@@ -0,0 +1,14 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
return [
|
||||
'connector' => 'Sync'
|
||||
];
|
||||
36
vendor/topthink/think-queue/src/queue/CallQueuedHandler.php
vendored
Normal file
36
vendor/topthink/think-queue/src/queue/CallQueuedHandler.php
vendored
Normal file
@@ -0,0 +1,36 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
class CallQueuedHandler
|
||||
{
|
||||
|
||||
public function call(Job $job, array $data)
|
||||
{
|
||||
$command = unserialize($data['command']);
|
||||
|
||||
call_user_func([$command, 'handle']);
|
||||
|
||||
if (!$job->isDeletedOrReleased()) {
|
||||
$job->delete();
|
||||
}
|
||||
}
|
||||
|
||||
public function failed(array $data)
|
||||
{
|
||||
$command = unserialize($data['command']);
|
||||
|
||||
if (method_exists($command, 'failed')) {
|
||||
$command->failed();
|
||||
}
|
||||
}
|
||||
}
|
||||
69
vendor/topthink/think-queue/src/queue/Connector.php
vendored
Normal file
69
vendor/topthink/think-queue/src/queue/Connector.php
vendored
Normal file
@@ -0,0 +1,69 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
use InvalidArgumentException;
|
||||
|
||||
abstract class Connector
|
||||
{
|
||||
protected $options = [];
|
||||
|
||||
abstract public function push($job, $data = '', $queue = null);
|
||||
|
||||
abstract public function later($delay, $job, $data = '', $queue = null);
|
||||
|
||||
abstract public function pop($queue = null);
|
||||
|
||||
public function marshal()
|
||||
{
|
||||
throw new \RuntimeException('pop queues not support for this type');
|
||||
}
|
||||
|
||||
protected function createPayload($job, $data = '', $queue = null)
|
||||
{
|
||||
if (is_object($job)) {
|
||||
$payload = json_encode([
|
||||
'job' => 'think\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;
|
||||
}
|
||||
|
||||
protected function createPlainPayload($job, $data)
|
||||
{
|
||||
return ['job' => $job, 'data' => $data];
|
||||
}
|
||||
|
||||
protected 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;
|
||||
}
|
||||
}
|
||||
213
vendor/topthink/think-queue/src/queue/Job.php
vendored
Normal file
213
vendor/topthink/think-queue/src/queue/Job.php
vendored
Normal file
@@ -0,0 +1,213 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
use DateTime;
|
||||
use think\App;
|
||||
|
||||
abstract class Job
|
||||
{
|
||||
|
||||
/**
|
||||
* The job handler instance.
|
||||
* @var mixed
|
||||
*/
|
||||
protected $instance;
|
||||
|
||||
/**
|
||||
* The name of the queue the job belongs to.
|
||||
* @var string
|
||||
*/
|
||||
protected $queue;
|
||||
|
||||
/**
|
||||
* Indicates if the job has been deleted.
|
||||
* @var bool
|
||||
*/
|
||||
protected $deleted = false;
|
||||
|
||||
/**
|
||||
* Indicates if the job has been released.
|
||||
* @var bool
|
||||
*/
|
||||
protected $released = false;
|
||||
|
||||
/**
|
||||
* Fire the job.
|
||||
* @return void
|
||||
*/
|
||||
abstract public function fire();
|
||||
|
||||
/**
|
||||
* Delete the job from the queue.
|
||||
* @return void
|
||||
*/
|
||||
public function delete()
|
||||
{
|
||||
$this->deleted = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if the job has been deleted.
|
||||
* @return bool
|
||||
*/
|
||||
public function isDeleted()
|
||||
{
|
||||
return $this->deleted;
|
||||
}
|
||||
|
||||
/**
|
||||
* Release the job back into the queue.
|
||||
* @param int $delay
|
||||
* @return void
|
||||
*/
|
||||
public function release($delay = 0)
|
||||
{
|
||||
$this->released = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if the job was released back into the queue.
|
||||
* @return bool
|
||||
*/
|
||||
public function isReleased()
|
||||
{
|
||||
return $this->released;
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if the job has been deleted or released.
|
||||
* @return bool
|
||||
*/
|
||||
public function isDeletedOrReleased()
|
||||
{
|
||||
return $this->isDeleted() || $this->isReleased();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of times the job has been attempted.
|
||||
* @return int
|
||||
*/
|
||||
abstract public function attempts();
|
||||
|
||||
/**
|
||||
* Get the raw body string for the job.
|
||||
* @return string
|
||||
*/
|
||||
abstract public function getRawBody();
|
||||
|
||||
/**
|
||||
* Resolve and fire the job handler method.
|
||||
* @param array $payload
|
||||
* @return void
|
||||
*/
|
||||
protected function resolveAndFire(array $payload)
|
||||
{
|
||||
list($class, $method) = $this->parseJob($payload['job']);
|
||||
|
||||
$this->instance = $this->resolve($class);
|
||||
if ($this->instance) {
|
||||
$this->instance->{$method}($this, $payload['data']);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse the job declaration into class and method.
|
||||
* @param string $job
|
||||
* @return array
|
||||
*/
|
||||
protected function parseJob($job)
|
||||
{
|
||||
$segments = explode('@', $job);
|
||||
|
||||
return count($segments) > 1 ? $segments : [$segments[0], 'fire'];
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the given job handler.
|
||||
* @param string $name
|
||||
* @return mixed
|
||||
*/
|
||||
protected function resolve($name)
|
||||
{
|
||||
if (strpos($name, '\\') === false) {
|
||||
|
||||
if (strpos($name, '/') === false) {
|
||||
$module = '';
|
||||
} else {
|
||||
list($module, $name) = explode('/', $name, 2);
|
||||
}
|
||||
|
||||
$name = App::$namespace . ($module ? '\\' . strtolower($module) : '') . '\\job\\' . $name;
|
||||
}
|
||||
if (class_exists($name)) {
|
||||
return new $name();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Call the failed method on the job instance.
|
||||
* @return void
|
||||
*/
|
||||
public function failed()
|
||||
{
|
||||
$payload = json_decode($this->getRawBody(), true);
|
||||
|
||||
list($class, $method) = $this->parseJob($payload['job']);
|
||||
|
||||
$this->instance = $this->resolve($class);
|
||||
if ($this->instance && method_exists($this->instance, 'failed')) {
|
||||
$this->instance->failed($payload['data']);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate the number of seconds with the given delay.
|
||||
* @param \DateTime|int $delay
|
||||
* @return int
|
||||
*/
|
||||
protected function getSeconds($delay)
|
||||
{
|
||||
if ($delay instanceof DateTime) {
|
||||
return max(0, $delay->getTimestamp() - $this->getTime());
|
||||
}
|
||||
|
||||
return (int) $delay;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current system time.
|
||||
* @return int
|
||||
*/
|
||||
protected function getTime()
|
||||
{
|
||||
return time();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the name of the queued job class.
|
||||
* @return string
|
||||
*/
|
||||
public function getName()
|
||||
{
|
||||
return json_decode($this->getRawBody(), true)['job'];
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the name of the queue the job belongs to.
|
||||
* @return string
|
||||
*/
|
||||
public function getQueue()
|
||||
{
|
||||
return $this->queue;
|
||||
}
|
||||
}
|
||||
164
vendor/topthink/think-queue/src/queue/Listener.php
vendored
Normal file
164
vendor/topthink/think-queue/src/queue/Listener.php
vendored
Normal file
@@ -0,0 +1,164 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
use Closure;
|
||||
use think\Process;
|
||||
|
||||
class Listener
|
||||
{
|
||||
|
||||
/**
|
||||
* @var string
|
||||
*/
|
||||
protected $commandPath;
|
||||
|
||||
/**
|
||||
* @var int
|
||||
*/
|
||||
protected $sleep = 3;
|
||||
|
||||
/**
|
||||
* @var int
|
||||
*/
|
||||
protected $maxTries = 0;
|
||||
|
||||
/**
|
||||
* @var string
|
||||
*/
|
||||
protected $workerCommand;
|
||||
|
||||
/**
|
||||
* @var \Closure|null
|
||||
*/
|
||||
protected $outputHandler;
|
||||
|
||||
/**
|
||||
* @param string $commandPath
|
||||
*/
|
||||
public function __construct($commandPath)
|
||||
{
|
||||
$this->commandPath = $commandPath;
|
||||
$this->workerCommand =
|
||||
'"' . PHP_BINARY . '" think queue:work --queue="%s" --delay=%s --memory=%s --sleep=%s --tries=%s';
|
||||
}
|
||||
|
||||
/**
|
||||
* @param string $queue
|
||||
* @param string $delay
|
||||
* @param string $memory
|
||||
* @param int $timeout
|
||||
* @return void
|
||||
*/
|
||||
public function listen($queue, $delay, $memory, $timeout = 60)
|
||||
{
|
||||
$process = $this->makeProcess($queue, $delay, $memory, $timeout);
|
||||
|
||||
while (true) {
|
||||
$this->runProcess($process, $memory);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param \Think\Process $process
|
||||
* @param int $memory
|
||||
*/
|
||||
public function runProcess(Process $process, $memory)
|
||||
{
|
||||
$process->run(function ($type, $line) {
|
||||
$this->handleWorkerOutput($type, $line);
|
||||
});
|
||||
|
||||
if ($this->memoryExceeded($memory)) {
|
||||
$this->stop();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param string $queue
|
||||
* @param int $delay
|
||||
* @param int $memory
|
||||
* @param int $timeout
|
||||
* @return \think\Process
|
||||
*/
|
||||
public function makeProcess($queue, $delay, $memory, $timeout)
|
||||
{
|
||||
$string = $this->workerCommand;
|
||||
$command = sprintf($string, $queue, $delay, $memory, $this->sleep, $this->maxTries);
|
||||
|
||||
return new Process($command, $this->commandPath, null, null, $timeout);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param int $type
|
||||
* @param string $line
|
||||
* @return void
|
||||
*/
|
||||
protected function handleWorkerOutput($type, $line)
|
||||
{
|
||||
if (isset($this->outputHandler)) {
|
||||
call_user_func($this->outputHandler, $type, $line);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param int $memoryLimit
|
||||
* @return bool
|
||||
*/
|
||||
public function memoryExceeded($memoryLimit)
|
||||
{
|
||||
return (memory_get_usage() / 1024 / 1024) >= $memoryLimit;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return void
|
||||
*/
|
||||
public function stop()
|
||||
{
|
||||
die;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param \Closure $outputHandler
|
||||
* @return void
|
||||
*/
|
||||
public function setOutputHandler(Closure $outputHandler)
|
||||
{
|
||||
$this->outputHandler = $outputHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return int
|
||||
*/
|
||||
public function getSleep()
|
||||
{
|
||||
return $this->sleep;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param int $sleep
|
||||
* @return void
|
||||
*/
|
||||
public function setSleep($sleep)
|
||||
{
|
||||
$this->sleep = $sleep;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param int $tries
|
||||
* @return void
|
||||
*/
|
||||
public function setMaxTries($tries)
|
||||
{
|
||||
$this->maxTries = $tries;
|
||||
}
|
||||
}
|
||||
46
vendor/topthink/think-queue/src/queue/Queueable.php
vendored
Normal file
46
vendor/topthink/think-queue/src/queue/Queueable.php
vendored
Normal file
@@ -0,0 +1,46 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
trait Queueable
|
||||
{
|
||||
|
||||
/** @var string 队列名称 */
|
||||
public $queue;
|
||||
|
||||
/** @var integer 延迟时间 */
|
||||
public $delay;
|
||||
|
||||
/**
|
||||
* 设置队列名
|
||||
* @param $queue
|
||||
* @return $this
|
||||
*/
|
||||
public function queue($queue)
|
||||
{
|
||||
$this->queue = $queue;
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置延迟时间
|
||||
* @param $delay
|
||||
* @return $this
|
||||
*/
|
||||
public function delay($delay)
|
||||
{
|
||||
$this->delay = $delay;
|
||||
|
||||
return $this;
|
||||
}
|
||||
}
|
||||
17
vendor/topthink/think-queue/src/queue/ShouldQueue.php
vendored
Normal file
17
vendor/topthink/think-queue/src/queue/ShouldQueue.php
vendored
Normal file
@@ -0,0 +1,17 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
interface ShouldQueue
|
||||
{
|
||||
|
||||
}
|
||||
119
vendor/topthink/think-queue/src/queue/Worker.php
vendored
Normal file
119
vendor/topthink/think-queue/src/queue/Worker.php
vendored
Normal file
@@ -0,0 +1,119 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue;
|
||||
|
||||
use Exception;
|
||||
use think\Hook;
|
||||
use think\Queue;
|
||||
|
||||
class Worker
|
||||
{
|
||||
|
||||
/**
|
||||
* 执行下个任务
|
||||
* @param string $queue
|
||||
* @param int $delay
|
||||
* @param int $sleep
|
||||
* @param int $maxTries
|
||||
* @return array
|
||||
*/
|
||||
public function pop($queue = null, $delay = 0, $sleep = 3, $maxTries = 0)
|
||||
{
|
||||
|
||||
$job = $this->getNextJob($queue);
|
||||
|
||||
if (!is_null($job)) {
|
||||
Hook::listen('worker_before_process', $queue);
|
||||
return $this->process($job, $maxTries, $delay);
|
||||
}
|
||||
|
||||
Hook::listen('worker_before_sleep', $queue);
|
||||
$this->sleep($sleep);
|
||||
|
||||
return ['job' => null, 'failed' => false];
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取下个任务
|
||||
* @param string $queue
|
||||
* @return Job
|
||||
*/
|
||||
protected function getNextJob($queue)
|
||||
{
|
||||
if (is_null($queue)) {
|
||||
return Queue::pop();
|
||||
}
|
||||
|
||||
foreach (explode(',', $queue) as $queue) {
|
||||
if (!is_null($job = Queue::pop($queue))) {
|
||||
return $job;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a given job from the queue.
|
||||
* @param \think\queue\Job $job
|
||||
* @param int $maxTries
|
||||
* @param int $delay
|
||||
* @return array
|
||||
* @throws Exception
|
||||
*/
|
||||
public function process(Job $job, $maxTries = 0, $delay = 0)
|
||||
{
|
||||
if ($maxTries > 0 && $job->attempts() > $maxTries) {
|
||||
return $this->logFailedJob($job);
|
||||
}
|
||||
|
||||
try {
|
||||
$job->fire();
|
||||
|
||||
return ['job' => $job, 'failed' => false];
|
||||
} catch (Exception $e) {
|
||||
if (!$job->isDeleted()) {
|
||||
$job->release($delay);
|
||||
}
|
||||
|
||||
throw $e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Log a failed job into storage.
|
||||
* @param \Think\Queue\Job $job
|
||||
* @return array
|
||||
*/
|
||||
protected function logFailedJob(Job $job)
|
||||
{
|
||||
if (!$job->isDeleted()) {
|
||||
try {
|
||||
$job->delete();
|
||||
$job->failed();
|
||||
} finally {
|
||||
Hook::listen('queue_failed', $job);
|
||||
}
|
||||
}
|
||||
|
||||
return ['job' => $job, 'failed' => true];
|
||||
}
|
||||
|
||||
/**
|
||||
* Sleep the script for a given number of seconds.
|
||||
* @param int $seconds
|
||||
* @return void
|
||||
*/
|
||||
public function sleep($seconds)
|
||||
{
|
||||
sleep($seconds);
|
||||
}
|
||||
|
||||
}
|
||||
65
vendor/topthink/think-queue/src/queue/command/Listen.php
vendored
Normal file
65
vendor/topthink/think-queue/src/queue/command/Listen.php
vendored
Normal file
@@ -0,0 +1,65 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\command;
|
||||
|
||||
use think\console\Command;
|
||||
use think\console\Input;
|
||||
use think\console\input\Option;
|
||||
use think\console\Output;
|
||||
use think\queue\Listener;
|
||||
|
||||
class Listen extends Command
|
||||
{
|
||||
/** @var Listener */
|
||||
protected $listener;
|
||||
|
||||
public function configure()
|
||||
{
|
||||
$this->setName('queue:listen')
|
||||
->addOption('queue', null, Option::VALUE_OPTIONAL, 'The queue to listen on', null)
|
||||
->addOption('delay', null, Option::VALUE_OPTIONAL, 'Amount of time to delay failed jobs', 0)
|
||||
->addOption('memory', null, Option::VALUE_OPTIONAL, 'The memory limit in megabytes', 128)
|
||||
->addOption('timeout', null, Option::VALUE_OPTIONAL, 'Seconds a job may run before timing out', 60)
|
||||
->addOption('sleep', null, Option::VALUE_OPTIONAL, 'Seconds to wait before checking queue for jobs', 3)
|
||||
->addOption('tries', null, Option::VALUE_OPTIONAL, 'Number of times to attempt a job before logging it failed', 0)
|
||||
->setDescription('Listen to a given queue');
|
||||
}
|
||||
|
||||
public function initialize(Input $input, Output $output)
|
||||
{
|
||||
$this->listener = new Listener($this->findCommandPath());
|
||||
$this->listener->setSleep($input->getOption('sleep'));
|
||||
$this->listener->setMaxTries($input->getOption('tries'));
|
||||
|
||||
$this->listener->setOutputHandler(function ($type, $line) use ($output) {
|
||||
$output->write($line);
|
||||
});
|
||||
}
|
||||
|
||||
public function execute(Input $input, Output $output)
|
||||
{
|
||||
$delay = $input->getOption('delay');
|
||||
|
||||
$memory = $input->getOption('memory');
|
||||
|
||||
$timeout = $input->getOption('timeout');
|
||||
|
||||
$queue = $input->getOption('queue') ?: 'default';
|
||||
|
||||
$this->listener->listen($queue, $delay, $memory, $timeout);
|
||||
}
|
||||
|
||||
protected function findCommandPath()
|
||||
{
|
||||
return defined('ROOT_PATH') ? ROOT_PATH : dirname($_SERVER['argv'][0]);
|
||||
}
|
||||
}
|
||||
31
vendor/topthink/think-queue/src/queue/command/Restart.php
vendored
Normal file
31
vendor/topthink/think-queue/src/queue/command/Restart.php
vendored
Normal file
@@ -0,0 +1,31 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\command;
|
||||
|
||||
use think\Cache;
|
||||
use think\console\Command;
|
||||
use think\console\Input;
|
||||
use think\console\Output;
|
||||
|
||||
class Restart extends Command
|
||||
{
|
||||
public function configure()
|
||||
{
|
||||
$this->setName('queue:restart')->setDescription('Restart queue worker daemons after their current job');
|
||||
}
|
||||
|
||||
public function execute(Input $input, Output $output)
|
||||
{
|
||||
Cache::set('think:queue:restart', time());
|
||||
$output->writeln("<info>Broadcasting queue restart signal.</info>");
|
||||
}
|
||||
}
|
||||
46
vendor/topthink/think-queue/src/queue/command/Subscribe.php
vendored
Normal file
46
vendor/topthink/think-queue/src/queue/command/Subscribe.php
vendored
Normal file
@@ -0,0 +1,46 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\command;
|
||||
|
||||
use think\console\Command;
|
||||
use think\console\Input;
|
||||
use think\console\input\Argument;
|
||||
use think\console\input\Option;
|
||||
use think\console\Output;
|
||||
use think\Queue;
|
||||
use think\Url;
|
||||
|
||||
class Subscribe extends Command
|
||||
{
|
||||
public function configure()
|
||||
{
|
||||
$this->setName('queue:subscribe')
|
||||
->setDescription('Subscribe a URL to an push queue')
|
||||
->addArgument('name', Argument::REQUIRED, 'name')
|
||||
->addArgument('url', Argument::REQUIRED, 'The URL to be subscribed.')
|
||||
->addArgument('queue', Argument::OPTIONAL, 'The URL to be subscribed.')
|
||||
->addOption('option', null, Option::VALUE_IS_ARRAY | Option::VALUE_OPTIONAL, 'the options');
|
||||
}
|
||||
|
||||
public function execute(Input $input, Output $output)
|
||||
{
|
||||
|
||||
$url = $input->getArgument('url');
|
||||
if (!preg_match('/^https?:\/\//', $url)) {
|
||||
$url = Url::build($url);
|
||||
}
|
||||
|
||||
Queue::subscribe($input->getArgument('name'), $url, $input->getArgument('queue'), $input->getOption('option'));
|
||||
|
||||
$output->write('<info>Queue subscriber added:</info> <comment>' . $input->getArgument('url') . '</comment>');
|
||||
}
|
||||
}
|
||||
210
vendor/topthink/think-queue/src/queue/command/Work.php
vendored
Normal file
210
vendor/topthink/think-queue/src/queue/command/Work.php
vendored
Normal file
@@ -0,0 +1,210 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
namespace think\queue\command;
|
||||
|
||||
use think\Config;
|
||||
use think\console\Command;
|
||||
use think\console\Input;
|
||||
use think\console\input\Option;
|
||||
use think\console\Output;
|
||||
use think\Hook;
|
||||
use think\queue\Job;
|
||||
use think\queue\Worker;
|
||||
use Exception;
|
||||
use Throwable;
|
||||
use think\Cache;
|
||||
use think\exception\Handle;
|
||||
use think\exception\ThrowableError;
|
||||
|
||||
class Work extends Command
|
||||
{
|
||||
|
||||
/**
|
||||
* The queue worker instance.
|
||||
* @var \think\queue\Worker
|
||||
*/
|
||||
protected $worker;
|
||||
|
||||
protected function initialize(Input $input, Output $output)
|
||||
{
|
||||
$this->worker = new Worker();
|
||||
}
|
||||
|
||||
protected function configure()
|
||||
{
|
||||
$this->setName('queue:work')
|
||||
->addOption('queue', null, Option::VALUE_OPTIONAL, 'The queue to listen on')
|
||||
->addOption('daemon', null, Option::VALUE_NONE, 'Run the worker in daemon mode')
|
||||
->addOption('delay', null, Option::VALUE_OPTIONAL, 'Amount of time to delay failed jobs', 0)
|
||||
->addOption('force', null, Option::VALUE_NONE, 'Force the worker to run even in maintenance mode')
|
||||
->addOption('memory', null, Option::VALUE_OPTIONAL, 'The memory limit in megabytes', 128)
|
||||
->addOption('sleep', null, Option::VALUE_OPTIONAL, 'Number of seconds to sleep when no job is available', 3)
|
||||
->addOption('tries', null, Option::VALUE_OPTIONAL, 'Number of times to attempt a job before logging it failed', 0)
|
||||
->setDescription('Process the next job on a queue');
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the console command.
|
||||
* @param Input $input
|
||||
* @param Output $output
|
||||
* @return int|null|void
|
||||
*/
|
||||
public function execute(Input $input, Output $output)
|
||||
{
|
||||
$queue = $input->getOption('queue');
|
||||
|
||||
$delay = $input->getOption('delay');
|
||||
|
||||
$memory = $input->getOption('memory');
|
||||
|
||||
if ($input->getOption('daemon')) {
|
||||
Hook::listen('worker_daemon_start', $queue);
|
||||
$this->daemon(
|
||||
$queue, $delay, $memory,
|
||||
$input->getOption('sleep'), $input->getOption('tries')
|
||||
);
|
||||
} else {
|
||||
$response = $this->worker->pop($queue, $delay, $input->getOption('sleep'), $input->getOption('tries'));
|
||||
$this->output($response);
|
||||
}
|
||||
}
|
||||
|
||||
protected function output($response)
|
||||
{
|
||||
if (!is_null($response['job'])) {
|
||||
/** @var Job $job */
|
||||
$job = $response['job'];
|
||||
if ($response['failed']) {
|
||||
$this->output->writeln('<error>Failed:</error> ' . $job->getName());
|
||||
} else {
|
||||
$this->output->writeln('<info>Processed:</info> ' . $job->getName());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动一个守护进程执行任务.
|
||||
*
|
||||
* @param string $queue
|
||||
* @param int $delay
|
||||
* @param int $memory
|
||||
* @param int $sleep
|
||||
* @param int $maxTries
|
||||
* @return array
|
||||
*/
|
||||
protected function daemon($queue = null, $delay = 0, $memory = 128, $sleep = 3, $maxTries = 0)
|
||||
{
|
||||
$lastRestart = $this->getTimestampOfLastQueueRestart();
|
||||
|
||||
while (true) {
|
||||
$this->runNextJobForDaemon(
|
||||
$queue, $delay, $sleep, $maxTries
|
||||
);
|
||||
|
||||
if ( $this->memoryExceeded($memory) ) {
|
||||
Hook::listen('worker_memory_exceeded', $queue);
|
||||
$this->stop();
|
||||
}
|
||||
|
||||
if ( $this->queueShouldRestart($lastRestart) ) {
|
||||
Hook::listen('worker_queue_restart', $queue);
|
||||
$this->stop();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 以守护进程的方式执行下个任务.
|
||||
*
|
||||
* @param string $queue
|
||||
* @param int $delay
|
||||
* @param int $sleep
|
||||
* @param int $maxTries
|
||||
* @return void
|
||||
*/
|
||||
protected function runNextJobForDaemon($queue, $delay, $sleep, $maxTries)
|
||||
{
|
||||
try {
|
||||
$response = $this->worker->pop($queue, $delay, $sleep, $maxTries);
|
||||
|
||||
$this->output($response);
|
||||
} catch (Exception $e) {
|
||||
$this->getExceptionHandler()->report($e);
|
||||
} catch (Throwable $e) {
|
||||
$this->getExceptionHandler()->report(new ThrowableError($e));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取上次重启守护进程的时间
|
||||
*
|
||||
* @return int|null
|
||||
*/
|
||||
protected function getTimestampOfLastQueueRestart()
|
||||
{
|
||||
return Cache::get('think:queue:restart');
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查是否要重启守护进程
|
||||
*
|
||||
* @param int|null $lastRestart
|
||||
* @return bool
|
||||
*/
|
||||
protected function queueShouldRestart($lastRestart)
|
||||
{
|
||||
return $this->getTimestampOfLastQueueRestart() != $lastRestart;
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查内存是否超出
|
||||
* @param int $memoryLimit
|
||||
* @return bool
|
||||
*/
|
||||
protected function memoryExceeded($memoryLimit)
|
||||
{
|
||||
return (memory_get_usage() / 1024 / 1024) >= $memoryLimit;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取异常处理实例
|
||||
*
|
||||
* @return \think\exception\Handle
|
||||
*/
|
||||
protected function getExceptionHandler()
|
||||
{
|
||||
static $handle;
|
||||
|
||||
if (!$handle) {
|
||||
|
||||
if ($class = Config::get('exception_handle')) {
|
||||
if (class_exists($class) && is_subclass_of($class, "\\think\\exception\\Handle")) {
|
||||
$handle = new $class;
|
||||
}
|
||||
}
|
||||
if (!$handle) {
|
||||
$handle = new Handle();
|
||||
}
|
||||
}
|
||||
|
||||
return $handle;
|
||||
}
|
||||
|
||||
/**
|
||||
* 停止执行任务的守护进程.
|
||||
* @return void
|
||||
*/
|
||||
public function stop()
|
||||
{
|
||||
die;
|
||||
}
|
||||
|
||||
}
|
||||
171
vendor/topthink/think-queue/src/queue/connector/Database.php
vendored
Normal file
171
vendor/topthink/think-queue/src/queue/connector/Database.php
vendored
Normal file
@@ -0,0 +1,171 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\connector;
|
||||
|
||||
use think\Db;
|
||||
use think\queue\Connector;
|
||||
use think\queue\job\Database as DatabaseJob;
|
||||
|
||||
class Database extends Connector
|
||||
{
|
||||
protected $db;
|
||||
|
||||
protected $options = [
|
||||
'expire' => 60,
|
||||
'default' => 'default',
|
||||
'table' => 'jobs',
|
||||
'dsn' => []
|
||||
];
|
||||
|
||||
public function __construct($options)
|
||||
{
|
||||
if (!empty($options)) {
|
||||
$this->options = array_merge($this->options, $options);
|
||||
}
|
||||
|
||||
$this->db = Db::connect($this->options['dsn']);
|
||||
}
|
||||
|
||||
public function push($job, $data = '', $queue = null)
|
||||
{
|
||||
return $this->pushToDatabase(0, $queue, $this->createPayload($job, $data));
|
||||
}
|
||||
|
||||
public function later($delay, $job, $data = '', $queue = null)
|
||||
{
|
||||
return $this->pushToDatabase($delay, $queue, $this->createPayload($job, $data));
|
||||
}
|
||||
|
||||
public function pop($queue = null)
|
||||
{
|
||||
$queue = $this->getQueue($queue);
|
||||
|
||||
if (!is_null($this->options['expire'])) {
|
||||
$this->releaseJobsThatHaveBeenReservedTooLong($queue);
|
||||
}
|
||||
|
||||
if ($job = $this->getNextAvailableJob($queue)) {
|
||||
$this->markJobAsReserved($job->id);
|
||||
|
||||
$this->db->commit();
|
||||
|
||||
return new DatabaseJob($this, $job, $queue);
|
||||
}
|
||||
|
||||
$this->db->commit();
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新发布任务
|
||||
* @param string $queue
|
||||
* @param \StdClass $job
|
||||
* @param int $delay
|
||||
* @return mixed
|
||||
*/
|
||||
public function release($queue, $job, $delay)
|
||||
{
|
||||
return $this->pushToDatabase($delay, $queue, $job->payload, $job->attempts);
|
||||
}
|
||||
|
||||
/**
|
||||
* Push a raw payload to the database with a given delay.
|
||||
*
|
||||
* @param \DateTime|int $delay
|
||||
* @param string|null $queue
|
||||
* @param string $payload
|
||||
* @param int $attempts
|
||||
* @return mixed
|
||||
*/
|
||||
protected function pushToDatabase($delay, $queue, $payload, $attempts = 0)
|
||||
{
|
||||
return $this->db->name($this->options['table'])->insert([
|
||||
'queue' => $this->getQueue($queue),
|
||||
'payload' => $payload,
|
||||
'attempts' => $attempts,
|
||||
'reserved' => 0,
|
||||
'reserved_at' => null,
|
||||
'available_at' => time() + $delay,
|
||||
'created_at' => time()
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取下个有效任务
|
||||
*
|
||||
* @param string|null $queue
|
||||
* @return \StdClass|null
|
||||
*/
|
||||
protected function getNextAvailableJob($queue)
|
||||
{
|
||||
$this->db->startTrans();
|
||||
|
||||
$job = $this->db->name($this->options['table'])
|
||||
->lock(true)
|
||||
->where('queue', $this->getQueue($queue))
|
||||
->where('reserved', 0)
|
||||
->where('available_at', '<=', time())
|
||||
->order('id', 'asc')
|
||||
->find();
|
||||
|
||||
return $job ? (object) $job : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记任务正在执行.
|
||||
*
|
||||
* @param string $id
|
||||
* @return void
|
||||
*/
|
||||
protected function markJobAsReserved($id)
|
||||
{
|
||||
$this->db->name($this->options['table'])->where('id', $id)->update([
|
||||
'reserved' => 1,
|
||||
'reserved_at' => time()
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新发布超时的任务
|
||||
*
|
||||
* @param string $queue
|
||||
* @return void
|
||||
*/
|
||||
protected function releaseJobsThatHaveBeenReservedTooLong($queue)
|
||||
{
|
||||
$expired = time() - $this->options['expire'];
|
||||
|
||||
$this->db->name($this->options['table'])
|
||||
->where('queue', $this->getQueue($queue))
|
||||
->where('reserved', 1)
|
||||
->where('reserved_at', '<=', $expired)
|
||||
->update([
|
||||
'reserved' => 0,
|
||||
'reserved_at' => null,
|
||||
'attempts' => ['inc', 1]
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除任务
|
||||
* @param string $id
|
||||
* @return void
|
||||
*/
|
||||
public function deleteReserved($id)
|
||||
{
|
||||
$this->db->name($this->options['table'])->delete($id);
|
||||
}
|
||||
|
||||
protected function getQueue($queue)
|
||||
{
|
||||
return $queue ?: $this->options['default'];
|
||||
}
|
||||
}
|
||||
236
vendor/topthink/think-queue/src/queue/connector/Redis.php
vendored
Normal file
236
vendor/topthink/think-queue/src/queue/connector/Redis.php
vendored
Normal file
@@ -0,0 +1,236 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\connector;
|
||||
|
||||
use Exception;
|
||||
use think\helper\Str;
|
||||
use think\queue\Connector;
|
||||
use think\queue\job\Redis as RedisJob;
|
||||
|
||||
class Redis extends Connector
|
||||
{
|
||||
/** @var \Redis */
|
||||
protected $redis;
|
||||
|
||||
protected $options = [
|
||||
'expire' => 60,
|
||||
'default' => 'default',
|
||||
'host' => '127.0.0.1',
|
||||
'port' => 6379,
|
||||
'password' => '',
|
||||
'select' => 0,
|
||||
'timeout' => 0,
|
||||
'persistent' => false
|
||||
];
|
||||
|
||||
public function __construct($options)
|
||||
{
|
||||
if (!extension_loaded('redis')) {
|
||||
throw new Exception('redis扩展未安装');
|
||||
}
|
||||
if (!empty($options)) {
|
||||
$this->options = array_merge($this->options, $options);
|
||||
}
|
||||
|
||||
$func = $this->options['persistent'] ? 'pconnect' : 'connect';
|
||||
$this->redis = new \Redis;
|
||||
$this->redis->$func($this->options['host'], $this->options['port'], $this->options['timeout']);
|
||||
|
||||
if ('' != $this->options['password']) {
|
||||
$this->redis->auth($this->options['password']);
|
||||
}
|
||||
|
||||
if (0 != $this->options['select']) {
|
||||
$this->redis->select($this->options['select']);
|
||||
}
|
||||
}
|
||||
|
||||
public function push($job, $data = '', $queue = null)
|
||||
{
|
||||
return $this->pushRaw($this->createPayload($job, $data), $queue);
|
||||
}
|
||||
|
||||
public function later($delay, $job, $data = '', $queue = null)
|
||||
{
|
||||
$payload = $this->createPayload($job, $data);
|
||||
|
||||
$this->redis->zAdd($this->getQueue($queue) . ':delayed', time() + $delay, $payload);
|
||||
}
|
||||
|
||||
public function pop($queue = null)
|
||||
{
|
||||
$original = $queue ?: $this->options['default'];
|
||||
|
||||
$queue = $this->getQueue($queue);
|
||||
|
||||
$this->migrateExpiredJobs($queue . ':delayed', $queue, false);
|
||||
|
||||
if (!is_null($this->options['expire'])) {
|
||||
$this->migrateExpiredJobs($queue . ':reserved', $queue);
|
||||
}
|
||||
|
||||
$job = $this->redis->lPop($queue);
|
||||
|
||||
if ($job !== false) {
|
||||
$this->redis->zAdd($queue . ':reserved', time() + $this->options['expire'], $job);
|
||||
|
||||
return new RedisJob($this, $job, $original);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新发布任务
|
||||
*
|
||||
* @param string $queue
|
||||
* @param string $payload
|
||||
* @param int $delay
|
||||
* @param int $attempts
|
||||
* @return void
|
||||
*/
|
||||
public function release($queue, $payload, $delay, $attempts)
|
||||
{
|
||||
$payload = $this->setMeta($payload, 'attempts', $attempts);
|
||||
|
||||
$this->redis->zAdd($this->getQueue($queue) . ':delayed', time() + $delay, $payload);
|
||||
}
|
||||
|
||||
public function pushRaw($payload, $queue = null)
|
||||
{
|
||||
$this->redis->rPush($this->getQueue($queue), $payload);
|
||||
|
||||
return json_decode($payload, true)['id'];
|
||||
}
|
||||
|
||||
protected function createPayload($job, $data = '', $queue = null)
|
||||
{
|
||||
$payload = $this->setMeta(
|
||||
parent::createPayload($job, $data), 'id', $this->getRandomId()
|
||||
);
|
||||
|
||||
return $this->setMeta($payload, 'attempts', 1);
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除任务
|
||||
*
|
||||
* @param string $queue
|
||||
* @param string $job
|
||||
* @return void
|
||||
*/
|
||||
public function deleteReserved($queue, $job)
|
||||
{
|
||||
$this->redis->zRem($this->getQueue($queue) . ':reserved', $job);
|
||||
}
|
||||
|
||||
/**
|
||||
* 移动延迟任务
|
||||
*
|
||||
* @param string $from
|
||||
* @param string $to
|
||||
* @param bool $attempt
|
||||
*/
|
||||
public function migrateExpiredJobs($from, $to, $attempt = true)
|
||||
{
|
||||
$this->redis->watch($from);
|
||||
|
||||
$jobs = $this->getExpiredJobs(
|
||||
$from, $time = time()
|
||||
);
|
||||
if (count($jobs) > 0) {
|
||||
$this->transaction(function () use ($from, $to, $time, $jobs, $attempt) {
|
||||
$this->removeExpiredJobs($from, $time);
|
||||
$this->pushExpiredJobsOntoNewQueue($to, $jobs, $attempt);
|
||||
});
|
||||
}
|
||||
$this->redis->unwatch();
|
||||
}
|
||||
|
||||
/**
|
||||
* redis事务
|
||||
* @param \Closure $closure
|
||||
*/
|
||||
protected function transaction(\Closure $closure)
|
||||
{
|
||||
$this->redis->multi();
|
||||
try {
|
||||
call_user_func($closure);
|
||||
if (!$this->redis->exec()) {
|
||||
$this->redis->discard();
|
||||
}
|
||||
} catch (Exception $e) {
|
||||
$this->redis->discard();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取所有到期任务
|
||||
*
|
||||
* @param string $from
|
||||
* @param int $time
|
||||
* @return array
|
||||
*/
|
||||
protected function getExpiredJobs($from, $time)
|
||||
{
|
||||
return $this->redis->zRangeByScore($from, '-inf', $time);
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除过期任务
|
||||
*
|
||||
* @param string $from
|
||||
* @param int $time
|
||||
* @return void
|
||||
*/
|
||||
protected function removeExpiredJobs($from, $time)
|
||||
{
|
||||
$this->redis->zRemRangeByScore($from, '-inf', $time);
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新发布到期任务
|
||||
*
|
||||
* @param string $to
|
||||
* @param array $jobs
|
||||
* @param boolean $attempt
|
||||
*/
|
||||
protected function pushExpiredJobsOntoNewQueue($to, $jobs, $attempt = true)
|
||||
{
|
||||
if ($attempt) {
|
||||
foreach ($jobs as &$job) {
|
||||
$attempts = json_decode($job, true)['attempts'];
|
||||
$job = $this->setMeta($job, 'attempts', $attempts + 1);
|
||||
}
|
||||
}
|
||||
call_user_func_array([$this->redis, 'rPush'], array_merge([$to], $jobs));
|
||||
}
|
||||
|
||||
/**
|
||||
* 随机id
|
||||
*
|
||||
* @return string
|
||||
*/
|
||||
protected function getRandomId()
|
||||
{
|
||||
return Str::random(32);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取队列名
|
||||
*
|
||||
* @param string|null $queue
|
||||
* @return string
|
||||
*/
|
||||
protected function getQueue($queue)
|
||||
{
|
||||
return 'queues:' . ($queue ?: $this->options['default']);
|
||||
}
|
||||
}
|
||||
57
vendor/topthink/think-queue/src/queue/connector/Sync.php
vendored
Normal file
57
vendor/topthink/think-queue/src/queue/connector/Sync.php
vendored
Normal file
@@ -0,0 +1,57 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\connector;
|
||||
|
||||
use Exception;
|
||||
use think\queue\Connector;
|
||||
use think\queue\job\Sync as SyncJob;
|
||||
use Throwable;
|
||||
|
||||
class Sync extends Connector
|
||||
{
|
||||
|
||||
public function push($job, $data = '', $queue = null)
|
||||
{
|
||||
$queueJob = $this->resolveJob($this->createPayload($job, $data, $queue));
|
||||
|
||||
try {
|
||||
set_time_limit(0);
|
||||
$queueJob->fire();
|
||||
} catch (Exception $e) {
|
||||
$queueJob->failed();
|
||||
|
||||
throw $e;
|
||||
} catch (Throwable $e) {
|
||||
$queueJob->failed();
|
||||
|
||||
throw $e;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
public function later($delay, $job, $data = '', $queue = null)
|
||||
{
|
||||
return $this->push($job, $data, $queue);
|
||||
}
|
||||
|
||||
public function pop($queue = null)
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
protected function resolveJob($payload)
|
||||
{
|
||||
return new SyncJob($payload);
|
||||
}
|
||||
|
||||
}
|
||||
225
vendor/topthink/think-queue/src/queue/connector/Topthink.php
vendored
Normal file
225
vendor/topthink/think-queue/src/queue/connector/Topthink.php
vendored
Normal file
@@ -0,0 +1,225 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\connector;
|
||||
|
||||
use think\exception\HttpException;
|
||||
use think\queue\Connector;
|
||||
use think\Request;
|
||||
use think\queue\job\Topthink as TopthinkJob;
|
||||
use think\Response;
|
||||
|
||||
class Topthink extends Connector
|
||||
{
|
||||
protected $options = [
|
||||
'token' => '',
|
||||
'project_id' => '',
|
||||
'protocol' => 'https',
|
||||
'host' => 'qns.topthink.com',
|
||||
'port' => 443,
|
||||
'api_version' => 1,
|
||||
'max_retries' => 3,
|
||||
'default' => 'default'
|
||||
];
|
||||
|
||||
/** @var Request */
|
||||
protected $request;
|
||||
|
||||
protected $url;
|
||||
|
||||
protected $curl = null;
|
||||
|
||||
protected $last_status;
|
||||
|
||||
protected $headers = [];
|
||||
|
||||
public function __construct($options)
|
||||
{
|
||||
if (!empty($options)) {
|
||||
$this->options = array_merge($this->options, $options);
|
||||
}
|
||||
|
||||
$this->url = "{$this->options['protocol']}://{$this->options['host']}:{$this->options['port']}/v{$this->options['api_version']}/";
|
||||
|
||||
$this->headers['Authorization'] = "Bearer {$this->options['token']}";
|
||||
|
||||
$this->request = Request::instance();
|
||||
}
|
||||
|
||||
public function push($job, $data = '', $queue = null)
|
||||
{
|
||||
return $this->pushRaw(0, $queue, $this->createPayload($job, $data));
|
||||
}
|
||||
|
||||
public function later($delay, $job, $data = '', $queue = null)
|
||||
{
|
||||
return $this->pushRaw($delay, $queue, $this->createPayload($job, $data));
|
||||
}
|
||||
|
||||
public function release($queue, $job, $delay)
|
||||
{
|
||||
return $this->pushRaw($delay, $queue, $job->payload, $job->attempts);
|
||||
}
|
||||
|
||||
public function marshal()
|
||||
{
|
||||
$job = new TopthinkJob($this, $this->marshalPushedJob(), $this->request->header('topthink-message-queue'));
|
||||
if ($this->request->header('topthink-message-status') == 'success') {
|
||||
$job->fire();
|
||||
} else {
|
||||
$job->failed();
|
||||
}
|
||||
return new Response('OK');
|
||||
}
|
||||
|
||||
public function pushRaw($delay, $queue, $payload, $attempts = 0)
|
||||
{
|
||||
$queue_name = $this->getQueue($queue);
|
||||
$queue = rawurlencode($queue_name);
|
||||
$url = "project/{$this->options['project_id']}/queue/{$queue}/message";
|
||||
$message = [
|
||||
'payload' => $payload,
|
||||
'attempts' => $attempts,
|
||||
'delay' => $delay
|
||||
];
|
||||
|
||||
return $this->apiCall('POST', $url, $message)->id;
|
||||
}
|
||||
|
||||
public function deleteMessage($queue, $id)
|
||||
{
|
||||
$queue = rawurlencode($queue);
|
||||
$url = "project/{$this->options['project_id']}/queue/{$queue}/message/{$id}";
|
||||
return $this->apiCall('DELETE', $url);
|
||||
}
|
||||
|
||||
protected function apiCall($type, $url, $params = [])
|
||||
{
|
||||
$url = "{$this->url}$url";
|
||||
|
||||
if ($this->curl == null) {
|
||||
$this->curl = curl_init();
|
||||
}
|
||||
|
||||
switch ($type = strtoupper($type)) {
|
||||
case 'DELETE':
|
||||
curl_setopt($this->curl, CURLOPT_URL, $url);
|
||||
curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
|
||||
curl_setopt($this->curl, CURLOPT_POSTFIELDS, json_encode($params));
|
||||
break;
|
||||
case 'PUT':
|
||||
curl_setopt($this->curl, CURLOPT_URL, $url);
|
||||
curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
|
||||
curl_setopt($this->curl, CURLOPT_POSTFIELDS, json_encode($params));
|
||||
break;
|
||||
case 'POST':
|
||||
curl_setopt($this->curl, CURLOPT_URL, $url);
|
||||
curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
|
||||
curl_setopt($this->curl, CURLOPT_POST, true);
|
||||
curl_setopt($this->curl, CURLOPT_POSTFIELDS, $params);
|
||||
break;
|
||||
case 'GET':
|
||||
curl_setopt($this->curl, CURLOPT_POSTFIELDS, null);
|
||||
curl_setopt($this->curl, CURLOPT_CUSTOMREQUEST, $type);
|
||||
curl_setopt($this->curl, CURLOPT_HTTPGET, true);
|
||||
$url .= '?' . http_build_query($params);
|
||||
curl_setopt($this->curl, CURLOPT_URL, $url);
|
||||
break;
|
||||
}
|
||||
|
||||
curl_setopt($this->curl, CURLOPT_SSL_VERIFYPEER, false);
|
||||
curl_setopt($this->curl, CURLOPT_RETURNTRANSFER, true);
|
||||
|
||||
$headers = [];
|
||||
foreach ($this->headers as $k => $v) {
|
||||
if ($k == 'Connection') {
|
||||
$v = 'Close';
|
||||
}
|
||||
$headers[] = "$k: $v";
|
||||
}
|
||||
|
||||
curl_setopt($this->curl, CURLOPT_HTTPHEADER, $headers);
|
||||
curl_setopt($this->curl, CURLOPT_CONNECTTIMEOUT, 10);
|
||||
|
||||
return $this->callWithRetries();
|
||||
}
|
||||
|
||||
protected function callWithRetries()
|
||||
{
|
||||
for ($retry = 0; $retry < $this->options['max_retries']; $retry++) {
|
||||
$out = curl_exec($this->curl);
|
||||
if ($out === false) {
|
||||
$this->reportHttpError(0, curl_error($this->curl));
|
||||
}
|
||||
$this->last_status = curl_getinfo($this->curl, CURLINFO_HTTP_CODE);
|
||||
|
||||
if ($this->last_status >= 200 && $this->last_status < 300) {
|
||||
return self::jsonDecode($out);
|
||||
} elseif ($this->last_status >= 500) {
|
||||
self::waitRandomInterval($retry);
|
||||
} else {
|
||||
$this->reportHttpError($this->last_status, $out);
|
||||
}
|
||||
}
|
||||
$this->reportHttpError($this->last_status, "Service unavailable");
|
||||
return;
|
||||
}
|
||||
|
||||
protected static function jsonDecode($response)
|
||||
{
|
||||
$data = json_decode($response);
|
||||
|
||||
$json_error = json_last_error();
|
||||
if ($json_error != JSON_ERROR_NONE) {
|
||||
throw new \RuntimeException($json_error);
|
||||
}
|
||||
|
||||
return $data;
|
||||
}
|
||||
|
||||
protected static function waitRandomInterval($retry)
|
||||
{
|
||||
$max_delay = pow(4, $retry) * 100 * 1000;
|
||||
usleep(rand(0, $max_delay));
|
||||
}
|
||||
|
||||
protected function reportHttpError($status, $text)
|
||||
{
|
||||
throw new HttpException($status, "http error: {$status} | {$text}");
|
||||
}
|
||||
|
||||
/**
|
||||
* Marshal out the pushed job and payload.
|
||||
*
|
||||
* @return object
|
||||
*/
|
||||
protected function marshalPushedJob()
|
||||
{
|
||||
return (object) [
|
||||
'id' => $this->request->header('topthink-message-id'),
|
||||
'payload' => $this->request->getContent(),
|
||||
'attempts' => $this->request->header('topthink-message-attempts')
|
||||
];
|
||||
}
|
||||
|
||||
public function __destruct()
|
||||
{
|
||||
if ($this->curl != null) {
|
||||
curl_close($this->curl);
|
||||
$this->curl = null;
|
||||
}
|
||||
}
|
||||
|
||||
public function pop($queue = null)
|
||||
{
|
||||
throw new \RuntimeException('pop queues not support for this type');
|
||||
}
|
||||
}
|
||||
88
vendor/topthink/think-queue/src/queue/job/Database.php
vendored
Normal file
88
vendor/topthink/think-queue/src/queue/job/Database.php
vendored
Normal file
@@ -0,0 +1,88 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
namespace think\queue\job;
|
||||
|
||||
use think\queue\Job;
|
||||
use think\queue\connector\Database as DatabaseQueue;
|
||||
|
||||
class Database extends Job
|
||||
{
|
||||
/**
|
||||
* The database queue instance.
|
||||
* @var DatabaseQueue
|
||||
*/
|
||||
protected $database;
|
||||
|
||||
/**
|
||||
* The database job payload.
|
||||
* @var Object
|
||||
*/
|
||||
protected $job;
|
||||
|
||||
public function __construct(DatabaseQueue $database, $job, $queue)
|
||||
{
|
||||
$this->job = $job;
|
||||
$this->queue = $queue;
|
||||
$this->database = $database;
|
||||
$this->job->attempts = $this->job->attempts + 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行任务
|
||||
* @return void
|
||||
*/
|
||||
public function fire()
|
||||
{
|
||||
$this->resolveAndFire(json_decode($this->job->payload, true));
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除任务
|
||||
* @return void
|
||||
*/
|
||||
public function delete()
|
||||
{
|
||||
parent::delete();
|
||||
$this->database->deleteReserved($this->job->id);
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新发布任务
|
||||
* @param int $delay
|
||||
* @return void
|
||||
*/
|
||||
public function release($delay = 0)
|
||||
{
|
||||
parent::release($delay);
|
||||
|
||||
$this->delete();
|
||||
|
||||
$this->database->release($this->queue, $this->job, $delay);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取当前任务尝试次数
|
||||
* @return int
|
||||
*/
|
||||
public function attempts()
|
||||
{
|
||||
return (int) $this->job->attempts;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the raw body string for the job.
|
||||
* @return string
|
||||
*/
|
||||
public function getRawBody()
|
||||
{
|
||||
return $this->job->payload;
|
||||
}
|
||||
}
|
||||
92
vendor/topthink/think-queue/src/queue/job/Redis.php
vendored
Normal file
92
vendor/topthink/think-queue/src/queue/job/Redis.php
vendored
Normal file
@@ -0,0 +1,92 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\job;
|
||||
|
||||
use think\queue\Job;
|
||||
use think\queue\connector\Redis as RedisQueue;
|
||||
|
||||
class Redis extends Job
|
||||
{
|
||||
|
||||
/**
|
||||
* The redis queue instance.
|
||||
* @var RedisQueue
|
||||
*/
|
||||
protected $redis;
|
||||
|
||||
/**
|
||||
* The database job payload.
|
||||
* @var Object
|
||||
*/
|
||||
protected $job;
|
||||
|
||||
public function __construct(RedisQueue $redis, $job, $queue)
|
||||
{
|
||||
$this->job = $job;
|
||||
$this->queue = $queue;
|
||||
$this->redis = $redis;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire the job.
|
||||
* @return void
|
||||
*/
|
||||
public function fire()
|
||||
{
|
||||
$this->resolveAndFire(json_decode($this->getRawBody(), true));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of times the job has been attempted.
|
||||
* @return int
|
||||
*/
|
||||
public function attempts()
|
||||
{
|
||||
return json_decode($this->job, true)['attempts'];
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the raw body string for the job.
|
||||
* @return string
|
||||
*/
|
||||
public function getRawBody()
|
||||
{
|
||||
return $this->job;
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除任务
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
public function delete()
|
||||
{
|
||||
parent::delete();
|
||||
|
||||
$this->redis->deleteReserved($this->queue, $this->job);
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新发布任务
|
||||
*
|
||||
* @param int $delay
|
||||
* @return void
|
||||
*/
|
||||
public function release($delay = 0)
|
||||
{
|
||||
parent::release($delay);
|
||||
|
||||
$this->delete();
|
||||
|
||||
$this->redis->release($this->queue, $this->job, $delay, $this->attempts() + 1);
|
||||
}
|
||||
}
|
||||
56
vendor/topthink/think-queue/src/queue/job/Sync.php
vendored
Normal file
56
vendor/topthink/think-queue/src/queue/job/Sync.php
vendored
Normal file
@@ -0,0 +1,56 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\job;
|
||||
|
||||
use think\queue\Job;
|
||||
|
||||
class Sync extends Job
|
||||
{
|
||||
/**
|
||||
* The queue message data.
|
||||
*
|
||||
* @var string
|
||||
*/
|
||||
protected $payload;
|
||||
|
||||
public function __construct($payload)
|
||||
{
|
||||
$this->payload = $payload;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire the job.
|
||||
* @return void
|
||||
*/
|
||||
public function fire()
|
||||
{
|
||||
$this->resolveAndFire(json_decode($this->payload, true));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of times the job has been attempted.
|
||||
* @return int
|
||||
*/
|
||||
public function attempts()
|
||||
{
|
||||
return 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the raw body string for the job.
|
||||
* @return string
|
||||
*/
|
||||
public function getRawBody()
|
||||
{
|
||||
return $this->payload;
|
||||
}
|
||||
}
|
||||
85
vendor/topthink/think-queue/src/queue/job/Topthink.php
vendored
Normal file
85
vendor/topthink/think-queue/src/queue/job/Topthink.php
vendored
Normal file
@@ -0,0 +1,85 @@
|
||||
<?php
|
||||
// +----------------------------------------------------------------------
|
||||
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
|
||||
// +----------------------------------------------------------------------
|
||||
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
|
||||
// +----------------------------------------------------------------------
|
||||
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
|
||||
// +----------------------------------------------------------------------
|
||||
// | Author: yunwuxin <448901948@qq.com>
|
||||
// +----------------------------------------------------------------------
|
||||
|
||||
namespace think\queue\job;
|
||||
|
||||
use think\queue\Job;
|
||||
use think\queue\connector\Topthink as TopthinkQueue;
|
||||
|
||||
class Topthink extends Job
|
||||
{
|
||||
|
||||
/**
|
||||
* The Iron queue instance.
|
||||
*
|
||||
* @var TopthinkQueue
|
||||
*/
|
||||
protected $topthink;
|
||||
|
||||
/**
|
||||
* The IronMQ message instance.
|
||||
*
|
||||
* @var object
|
||||
*/
|
||||
protected $job;
|
||||
|
||||
public function __construct(TopthinkQueue $topthink, $job, $queue)
|
||||
{
|
||||
$this->topthink = $topthink;
|
||||
$this->job = $job;
|
||||
$this->queue = $queue;
|
||||
$this->job->attempts = $this->job->attempts + 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire the job.
|
||||
* @return void
|
||||
*/
|
||||
public function fire()
|
||||
{
|
||||
$this->resolveAndFire(json_decode($this->job->payload, true));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of times the job has been attempted.
|
||||
* @return int
|
||||
*/
|
||||
public function attempts()
|
||||
{
|
||||
return (int) $this->job->attempts;
|
||||
}
|
||||
|
||||
public function delete()
|
||||
{
|
||||
parent::delete();
|
||||
|
||||
$this->topthink->deleteMessage($this->queue, $this->job->id);
|
||||
}
|
||||
|
||||
public function release($delay = 0)
|
||||
{
|
||||
parent::release($delay);
|
||||
|
||||
$this->delete();
|
||||
|
||||
$this->topthink->release($this->queue, $this->job, $delay);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the raw body string for the job.
|
||||
* @return string
|
||||
*/
|
||||
public function getRawBody()
|
||||
{
|
||||
return $this->job->payload;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user