thinkphp6.0消息队列配置database驱动

作者：七棵菜日期：2022-11-30 栏目：计算机分类：1 人气：234

### 安装

thinkphp6.0消息队列安装

```
composer install topthink/think-queue
```

最新版本为`^3.0`，安装好之后自动生成配置文件`config/queue.php`

```
return [
    'default'     =&gt; 'database',
    'connections' =&gt; [
        'sync'     =&gt; [
            'type' =&gt; 'sync',
        ],
        'database' =&gt; [
            'type'       =&gt; 'database',
            'queue'      =&gt; 'default',
            'table'      =&gt; 'jobs',
            'connection' =&gt; null,
        ],
        'redis'    =&gt; [
            'type'       =&gt; 'redis',
            'queue'      =&gt; 'default',
            'host'       =&gt; '127.0.0.1',
            'port'       =&gt; 6379,
            'password'   =&gt; '',
            'select'     =&gt; 0,
            'timeout'    =&gt; 0,
            'persistent' =&gt; false,
        ],
    ],
    'failed'      =&gt; [
        'type'  =&gt; 'none',
        'table' =&gt; 'failed_jobs',
    ],
];
```

但是如果把默认驱动改成`database`的话，源码有bug，主要问题有两个

- 构造函数参数$db的类型错误
- 数据操作对象单例模式造成`where`条件不断累加

自己尝试修复这一些bug，可以跑通，修复后的文件如下

```
&lt;?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 &lt;448901948@qq.com&gt;
// +----------------------------------------------------------------------

namespace think\queue\connector;

use Carbon\Carbon;
use stdClass;
use think\Db;
use think\db\ConnectionInterface;
use think\db\Query;
use think\queue\Connector;
use think\queue\InteractsWithTime;
use think\queue\job\Database as DatabaseJob;

class Database extends Connector
{

use InteractsWithTime;

protected $db;

/**
     * The database table that holds the jobs.
     *
     * @var string
     */
    protected $table;

/**
     * The name of the default queue.
     *
     * @var string
     */
    protected $default;

/**
     * The expiration time of a job.
     *
     * @var int|null
     */
    protected $retryAfter = 60;

public function __construct(Query $db, $table, $default = 'default', $retryAfter = 60)
    {
        $this-&gt;db         = $db;
        $this-&gt;table      = $table;
        $this-&gt;default    = $default;
        $this-&gt;retryAfter = $retryAfter;
    }

public static function __make(Db $db, $config)
    {
        $connection = $db-&gt;connect($config['connection'] ?? null);

return new self($connection, $config['table'], $config['queue'], $config['retry_after'] ?? 60);
    }

public function size($queue = null)
    {
        $this-&gt;db-&gt;name($this-&gt;table)
            -&gt;where('queue', $this-&gt;getQueue($queue))
            -&gt;count();
    }

public function push($job, $data = '', $queue = null)
    {
        return $this-&gt;pushToDatabase($queue, $this-&gt;createPayload($job, $data));
    }

public function pushRaw($payload, $queue = null, array $options = [])
    {
        return $this-&gt;pushToDatabase($queue, $payload);
    }

public function later($delay, $job, $data = '', $queue = null)
    {
        return $this-&gt;pushToDatabase($queue, $this-&gt;createPayload($job, $data), $delay);
    }

public function bulk($jobs, $data = '', $queue = null)
    {
        $queue = $this-&gt;getQueue($queue);

$availableAt = $this-&gt;availableAt();

return $this-&gt;db-&gt;name($this-&gt;table)-&gt;insertAll(collect((array) $jobs)-&gt;map(
            function ($job) use ($queue, $data, $availableAt) {
                return [
                    'queue'          =&gt; $queue,
                    'attempts'       =&gt; 0,
                    'reserve_time'   =&gt; null,
                    'available_time' =&gt; $availableAt,
                    'create_time'    =&gt; $this-&gt;currentTime(),
                    'payload'        =&gt; $this-&gt;createPayload($job, $data),
                ];
            }
        )-&gt;all());
    }

/**
     * 重新发布任务
     *
     * @param string $queue
     * @param StdClass $job
     * @param int $delay
     * @return mixed
     */
    public function release($queue, $job, $delay)
    {
        return $this-&gt;pushToDatabase($queue, $job-&gt;payload, $delay, $job-&gt;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($queue, $payload, $delay = 0, $attempts = 0)
    {
        return $this-&gt;db-&gt;name($this-&gt;table)-&gt;insertGetId([
            'queue'          =&gt; $this-&gt;getQueue($queue),
            'attempts'       =&gt; $attempts,
            'reserve_time'   =&gt; null,
            'available_time' =&gt; $this-&gt;availableAt($delay),
            'create_time'    =&gt; $this-&gt;currentTime(),
            'payload'        =&gt; $payload,
        ]);
    }

public function pop($queue = null)
    {
        $queue = $this-&gt;getQueue($queue);

return $this-&gt;db-&gt;transaction(function () use ($queue) {

if ($job = $this-&gt;getNextAvailableJob($queue)) {

$job = $this-&gt;markJobAsReserved($job);

return new DatabaseJob($this-&gt;app, $this, $job, $this-&gt;connection, $queue);
            }
        });
    }

/**
     * 获取下个有效任务
     *
     * @param string|null $queue
     * @return StdClass|null
     */
    protected function getNextAvailableJob($queue)
    {
        $db = clone $this-&gt;db;
        $job = $db-&gt;name($this-&gt;table)
            -&gt;lock(true)
            -&gt;where('queue', $this-&gt;getQueue($queue))
            -&gt;where(function (Query $query) {
                $query-&gt;where(function (Query $query) {
                    $query-&gt;whereNull('reserve_time')
                        -&gt;where('available_time', '&lt;=', $this-&gt;currentTime());
                });

//超时任务重试
                $expiration = Carbon::now()-&gt;subSeconds($this-&gt;retryAfter)-&gt;getTimestamp();

$query-&gt;whereOr(function (Query $query) use ($expiration) {
                    $query-&gt;where('reserve_time', '&lt;=', $expiration);
                });
            })
            -&gt;order('id', 'asc')
            -&gt;find();

return $job ? (object) $job : null;
    }

/**
     * 标记任务正在执行.
     *
     * @param stdClass $job
     * @return stdClass
     */
    protected function markJobAsReserved($job)
    {
        $db = clone $this-&gt;db;
        $db-&gt;name($this-&gt;table)
            -&gt;where('id', $job-&gt;id)
            -&gt;update([
                'reserve_time' =&gt; $job-&gt;reserve_time = $this-&gt;currentTime(),
                'attempts'     =&gt; ++$job-&gt;attempts,
            ]);
        return $job;
    }

/**
     * 删除任务
     *
     * @param string $id
     * @return void
     */
    public function deleteReserved($id)
    {
        $this-&gt;db-&gt;transaction(function () use ($id) {
            $db = clone $this-&gt;db;
            $db-&gt;name($this-&gt;table)-&gt;where('id', $id)-&gt;delete();
        });
    }

protected function getQueue($queue)
    {
        return $queue ?: $this-&gt;default;
    }
}

```

### 鸣谢

- [github](https://github.com/coolseven/notes/blob/master/thinkphp-queue/README.md)

标签： thinkphp6.0 消息队列 think-queue

上一篇：如何使用mvn命令导入依赖下一篇：apache 清除并且重新配置 X-FRAME-OPTIONS

thinkphp6.0消息队列配置database驱动

随便看看

留言