Laravel 7使用swoole+redis每秒处理任务

安装配置点击这里

redis 和 phpredis 安装完成后，在配置文件.env中配置QUEUE_CONNECTION=redis

创建定时任务类

namespace App\Jobs;

use App\Entities\Cron;
use Hhxsv5\LaravelS\Swoole\Timer\CronJob;
use Illuminate\Support\Facades\Log;
use Swoole\Coroutine;

class TestCron extends CronJob
{
    public function run()
    {
        // Swoole>=2.1 run()方法已自动创建了协程。
        Coroutine::sleep(1);
        // 查询数据库
        $lists = Cron::where([
            ['exe_time', '<=', time()],
            ['status', '=', 2],
            ['queue_status', '=', 1]
        ])->get();
        // 有数据放入队列中
        if(count($lists) > 0) {
            // 放入队列中
            foreach($lists as $k => $v) {
                TestCronJob::dispatch($v)->onConnection('redis')->onQueue('CronJobQueue');
                // 修改队列状态为队列中
                Cron::where(['id' => $v->id])->update(['queue_status' => 2]);
                Log::info(__METHOD__, Cron::find($v->id)->toArray());
            }
        }
    }
}

创建一个队列类

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;

class TestCronJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    // 队列数据
    protected $param;

    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct($param)
    {
        $this->param = $param;
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        // 判断请求方式，2 为get请求
        if ($this->param->type == 2) {
            // todo 业务逻辑
        }
    }
}

配置timer

// 在"config/laravels.php"注册定时任务类
[
    // ...
    'timer'          => [
        'enable' => env('LARAVELS_TIMER', true), // 启用Timer
        'jobs'   => [ // 注册的定时任务类列表
            // 启用LaravelScheduleJob来执行`php artisan schedule:run`，每分钟一次，替代Linux Crontab
            // \Hhxsv5\LaravelS\Illuminate\LaravelScheduleJob::class,
            // 两种配置参数的方式：
            // [\App\Jobs\Timer\TestCronJob::class, [1000, true]], // 注册时传入参数
            \App\Jobs\TestCron::class, // 重载对应的方法来返回参数
        ],
        'max_wait_time' => 5, // Reload时最大等待时间
        // 打开全局定时器开关：当多实例部署时，确保只有一个实例运行定时任务，此功能依赖 Redis，具体请看 https://learnku.com/docs/laravel/7.x/redis
        'global_lock'     => false,
        'global_lock_key' => config('app.name', 'Laravel'),
    ],
    // ...
];

测试

1. 首先启动laravals

$ php bin/laravels start
 _                               _  _____ 
| |                             | |/ ____|
| |     __ _ _ __ __ ___   _____| | (___  
| |    / _` | '__/ _` \ \ / / _ \ |\___ \ 
| |___| (_| | | | (_| |\ V /  __/ |____) |
|______\__,_|_|  \__,_| \_/ \___|_|_____/ 

Speed up your Laravel/Lumen
>>> Components
+---------------------------+---------+
| Component                 | Version |
+---------------------------+---------+
| PHP                       | 7.4.9   |
| Swoole                    | 4.5.3   |
| LaravelS                  | 3.7.8   |
| Laravel Framework [local] | 7.27.0  |
+---------------------------+---------+
>>> Protocols
+-----------+--------+-------------------+----------------+
| Protocol  | Status | Handler           | Listen At      |
+-----------+--------+-------------------+----------------+
| Main HTTP | On     | Laravel Framework | 127.0.0.1:5200 |
+-----------+--------+-------------------+----------------+
>>> Feedback: https://github.com/hhxsv5/laravel-s
[2020-09-07 16:17:59] [TRACE] Swoole is running, press Ctrl+C to quit.

2. 启动队列

$ php artisan queue:work redis  --queue=CronJobQueue
[2020-09-07 16:18:07][wxEp2oOyhnyMBGiCl309MLS7szGnz5gs] Processing: App\Jobs\TestCronJob
[2020-09-07 16:18:07][wxEp2oOyhnyMBGiCl309MLS7szGnz5gs] Processed:  App\Jobs\TestCronJob

3. 修改数据库数据
4. 查看日志

[2020-09-07 16:18:25] local.INFO: App\Jobs\TestCron::run {"id":67,"type":2,"exe_time":1599463122,"order_id":10000024,"status":2,"queue_status":2,"mch_order_id":"90806692346580992","args":"{\"ccode_url\":\"wxp\\/\\/f2f0XolMxgyUUdoYj10u2rzjKxuPmrkTemc5\",\"amount\":\"1\"}"} 
[2020-09-07 16:18:27] local.INFO: 请求上游接口===>https://www.test.com/test.php?project=test&id=10000024        返回状态===>未转账        结果:{"data":"999","code":200}  
[2020-09-07 16:18:27] local.INFO: 未转账,修改订单成功 [{"id":1,"mch_id":111,"ps_id":1,"mch_order_id":"2222","mch_notify_url":"http://notify.example.com","order_id":10000024,"order_status":4,"order_type":1,"ccode_url":"wxpay://1234jdkfjadjfkljadklf","amount":100,"create_time":"2020-09-04 16:18:11","update_time":"2020-09-07 16:18:27"}] 

5. 确认数据库是否修改成功

本作品采用《CC 协议》，转载必须注明作者和本文链接