队列支持

Author: yunwuxin

README.md

@@ -41,6 +41,38 @@ Route::get('static/:path', function (string $path) {
 
 2. 访问路由 `http://localhost/static/文件路径`
 
+## 队列支持
+
+使用方法见 [think-queue](https://github.com/top-think/think-queue)
+
+以下配置代替think-queue里的最后一步:`监听任务并执行`,无需另外起进程执行队列
+
+```php
+return [
+    // ...
+    'queue'      => [
+        'enable'  => true,
+        //键名是队列名称
+        'workers' => [
+            //下面参数是不设置时的默认配置
+            'default'            => [
+                'delay'      => 0,
+                'sleep'      => 3,
+                'tries'      => 0,
+                'timeout'    => 60,
+                'worker_num' => 1,
+            ],
+            //使用@符号后面可指定队列使用驱动
+            'default@connection' => [
+                //此处可不设置任何参数，使用上面的默认配置
+            ],
+        ],
+    ],
+    // ...
+];
+
+```
+
 ## 自定义worker
 监听`worker.init`事件 注入`Manager`对象，调用addWorker方法添加
 ~~~php

composer.json

@@ -31,6 +31,7 @@
     "require-dev": {
         "pestphp/pest": "^3.7",
         "guzzlehttp/guzzle": "^7.0",
+        "topthink/think-queue": "^3.0",
         "phpstan/phpstan": "^2.0"
     },
     "autoload-dev": {

phpstan.neon.dist renamed to phpstan.neon

@@ -12,3 +12,5 @@ parameters:
 		- '#Function config_path not found.#'
 		- '#Function public_path not found.#'
 		- '#Function json not found.#'
+		- '#While loop condition is always true#'
+		- '#Constant STUB_DIR not found.#'

phpunit.xml.dist renamed to phpunit.xml

@@ -3,6 +3,7 @@
 namespace think\worker;
 
 use think\worker\concerns\InteractsWithHttp;
+use think\worker\concerns\InteractsWithQueue;
 use think\worker\concerns\InteractsWithServer;
 use think\worker\concerns\WithApplication;
 use think\worker\concerns\WithContainer;
@@ -11,11 +12,13 @@ class Manager
 {
     use InteractsWithServer,
         InteractsWithHttp,
+        InteractsWithQueue,
         WithApplication,
         WithContainer;
 
     protected function initialize(): void
     {
         $this->prepareHttp();
+        $this->prepareQueue();
     }
 }

src/Manager.php

@@ -0,0 +1,80 @@
+<?php
+
+namespace think\worker\concerns;
+
+use think\helper\Arr;
+use think\queue\event\JobFailed;
+use think\queue\Worker;
+use Workerman\Timer;
+
+trait InteractsWithQueue
+{
+    protected function createQueueWorkers()
+    {
+        $workers = $this->getConfig('queue.workers', []);
+
+        foreach ($workers as $queue => $options) {
+
+            if (strpos($queue, '@') !== false) {
+                [$queue, $connection] = explode('@', $queue);
+            } else {
+                $connection = null;
+            }
+
+            $workerNum = Arr::get($options, 'worker_num', 1);
+
+            $this->addWorker(function (\think\worker\Worker $worker) use ($options, $connection, $queue) {
+                $delay   = Arr::get($options, 'delay', 0);
+                $sleep   = Arr::get($options, 'sleep', 3);
+                $tries   = Arr::get($options, 'tries', 0);
+                $timeout = Arr::get($options, 'timeout', 60);
+
+                $qWorker = $this->app->make(Worker::class);
+
+                while (true) {
+                    $timer = Timer::add($timeout, function () use ($worker) {
+                        $worker->stop();
+                    }, [], false);
+
+                    $this->runInSandbox(function () use ($connection, $queue, $delay, $sleep, $tries, $qWorker) {
+                        $qWorker->runNextJob($connection, $queue, $delay, $sleep, $tries);
+                    });
+
+                    Timer::del($timer);
+                }
+            }, "queue [$queue]", $workerNum);
+        }
+    }
+
+    public function prepareQueue()
+    {
+        if ($this->getConfig('queue.enable', false)) {
+            $this->listenForEvents();
+            $this->createQueueWorkers();
+        }
+    }
+
+    /**
+     * 注册事件
+     */
+    protected function listenForEvents()
+    {
+        $this->container->event->listen(JobFailed::class, function (JobFailed $event) {
+            $this->logFailedJob($event);
+        });
+    }
+
+    /**
+     * 记录失败任务
+     * @param JobFailed $event
+     */
+    protected function logFailedJob(JobFailed $event)
+    {
+        $this->container['queue.failer']->log(
+            $event->connection,
+            $event->job->getQueue(),
+            $event->job->getRawBody(),
+            $event->exception
+        );
+    }
+}

src/concerns/InteractsWithQueue.php

@@ -17,6 +17,11 @@
         'worker_num' => 2,
         'options'    => [],
     ],
+    //队列
+    'queue'      => [
+        'enable'  => false,
+        'workers' => [],
+    ],
     'hot_update' => [
         'enable'  => env('APP_DEBUG', false),
         'name'    => ['*.php'],

src/config/worker.php

@@ -1,42 +1,2 @@
 <?php
-
-/*
-|--------------------------------------------------------------------------
-| Test Case
-|--------------------------------------------------------------------------
-|
-| The closure you provide to your test functions is always bound to a specific PHPUnit test
-| case class. By default, that class is "PHPUnit\Framework\TestCase". Of course, you may
-| need to change it using the "uses()" function to bind a different classes or traits.
-|
-*/
-
-// uses(Tests\TestCase::class)->in('Feature');
-
-/*
-|--------------------------------------------------------------------------
-| Expectations
-|--------------------------------------------------------------------------
-|
-| When you're writing tests, you often need to check that values meet certain conditions. The
-| "expect()" function gives you access to a set of "expectations" methods that you can use
-| to assert different things. Of course, you may extend the Expectation API at any time.
-|
-*/
-
-
-/*
-|--------------------------------------------------------------------------
-| Functions
-|--------------------------------------------------------------------------
-|
-| While Pest is very powerful out-of-the-box, you may have some testing code specific to your
-| project that you don't want to repeat in every file. Here you can also expose helpers as
-| global functions to help you to reduce the number of lines of code in your test files.
-|
-*/
-
-function something()
-{
-    // ..
-}
+define('STUB_DIR', realpath(__DIR__ . '/stub'));

tests/Pest.php

@@ -1,12 +1,12 @@
 <?php
 
+use GuzzleHttp\Client;
 use GuzzleHttp\Cookie\CookieJar;
 use Symfony\Component\Process\Process;
-use GuzzleHttp\Client;
 
 $process = null;
 beforeAll(function () use (&$process) {
-    $process = new Process(['php', 'think', 'worker'], __DIR__ . '/stub/');
+    $process = new Process(['php', 'think', 'worker'], STUB_DIR);
     $process->start();
     $wait = 0;
 
@@ -25,15 +25,16 @@
 });
 
 beforeEach(function () {
-    $this->client = new Client([
+    $this->httpClient = new Client([
         'base_uri'    => 'http://127.0.0.1:8080',
         'cookies'     => true,
         'http_errors' => false,
+        'timeout'     => 1,
     ]);
 });
 
 it('callback route', function () {
-    $response = $this->client->get('/');
+    $response = $this->httpClient->get('/');
 
     expect($response->getStatusCode())
         ->toBe(200)
@@ -44,7 +45,7 @@
 it('controller route', function () {
     $jar = new CookieJar();
 
-    $response = $this->client->get('/test', ['cookies' => $jar]);
+    $response = $this->httpClient->get('/test', ['cookies' => $jar]);
 
     expect($response->getStatusCode())
         ->toBe(200)
@@ -59,7 +60,7 @@
     $data     = [
         'name' => 'think',
     ];
-    $response = $this->client->post('/json', [
+    $response = $this->httpClient->post('/json', [
         'json' => $data,
     ]);
 
@@ -70,14 +71,14 @@
 });
 
 it('put and delete request', function () {
-    $response = $this->client->put('/');
+    $response = $this->httpClient->put('/');
 
     expect($response->getStatusCode())
         ->toBe(200)
         ->and($response->getBody()->getContents())
         ->toBe('put');
 
-    $response = $this->client->delete('/');
+    $response = $this->httpClient->delete('/');
 
     expect($response->getStatusCode())
         ->toBe(200)
@@ -86,16 +87,16 @@
 });
 
 it('file response', function () {
-    $response = $this->client->get('/static/asset.txt');
+    $response = $this->httpClient->get('/static/asset.txt');
 
     expect($response->getStatusCode())
         ->toBe(200)
         ->and($response->getBody()->getContents())
-        ->toBe(file_get_contents(__DIR__ . '/stub/public/asset.txt'));
+        ->toBe(file_get_contents(STUB_DIR . '/public/asset.txt'));
 });
 
 it('hot update', function () {
-    $response = $this->client->get('/hot');
+    $response = $this->httpClient->get('/hot');
 
     expect($response->getStatusCode())
         ->toBe(404);
@@ -110,16 +111,16 @@
 });
 PHP;
 
-    file_put_contents(__DIR__ . '/stub/route/hot.php', $route);
+    file_put_contents(STUB_DIR . '/route/hot.php', $route);
 
     sleep(2);
 
-    $response = $this->client->get('/hot');
+    $response = $this->httpClient->get('/hot');
 
     expect($response->getStatusCode())
         ->toBe(200)
         ->and($response->getBody()->getContents())
         ->toBe('hot');
 })->after(function () {
-    @unlink(__DIR__ . '/stub/route/hot.php');
+    @unlink(STUB_DIR . '/route/hot.php');
 })->skipOnWindows();

tests/HttpTest.php renamed to tests/feature/HttpTest.php

@@ -11,12 +11,19 @@
 
 return [
     'http'       => [
-        'enable'     => true,
+        'enable'     => env('HTTP_ENABLE', true),
         'host'       => '0.0.0.0',
         'port'       => 8080,
         'worker_num' => 2,
         'options'    => [],
     ],
+    //队列
+    'queue'      => [
+        'enable'  => env('QUEUE_ENABLE', false),
+        'workers' => [
+            'default' => [],
+        ],
+    ],
     'hot_update' => [
         'enable'  => env('APP_DEBUG', false),
         'name'    => ['*.php'],