Database.php
3.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
<?php
// +----------------------------------------------------------------------
// | ThinkCMF [ WE CAN DO IT MORE SIMPLE ]
// +----------------------------------------------------------------------
// | Copyright (c) 2013-2018 http://www.thinkcmf.com All rights reserved.
// +----------------------------------------------------------------------
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
// +---------------------------------------------------------------------
// | Author: 老猫 <catmat@thinkcmf.com>
// +----------------------------------------------------------------------
/*
* 数据库表
CREATE TABLE `cmf_queue_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,
`reserve_time` int(10) unsigned DEFAULT NULL,
`available_time` int(10) unsigned NOT NULL,
`create_time` int(10) unsigned NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
*/
namespace cmf\queue\connector;
use think\queue\connector\Database as DataBaseConnector;
class Database extends DataBaseConnector
{
protected $options = [
'expire' => 60,
'default' => 'default',
'table' => 'queue_jobs',
'dsn' => []
];
/**
* 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,
'reserve_time' => null,
'available_time' => time() + $delay,
'create_time' => 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_time', '<=', 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,
'reserve_time' => 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('reserve_time', '<=', $expired)
->update([
'reserved' => 0,
'reserve_time' => null,
'attempts' => ['exp', 'attempts + 1']
]);
}
}