mirror of
https://github.com/hibiken/asynq.git
synced 2025-10-22 09:56:12 +08:00
chore(): 从内存里面读取任务
This commit is contained in:
32
inspector.go
32
inspector.go
@@ -15,14 +15,12 @@ import (
|
|||||||
"github.com/hibiken/asynq/internal/base"
|
"github.com/hibiken/asynq/internal/base"
|
||||||
"github.com/hibiken/asynq/internal/errors"
|
"github.com/hibiken/asynq/internal/errors"
|
||||||
"github.com/hibiken/asynq/internal/rdb"
|
"github.com/hibiken/asynq/internal/rdb"
|
||||||
"github.com/robfig/cron/v3"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// Inspector is a client interface to inspect and mutate the state of
|
// Inspector is a client interface to inspect and mutate the state of
|
||||||
// queues and tasks.
|
// queues and tasks.
|
||||||
type Inspector struct {
|
type Inspector struct {
|
||||||
rdb *rdb.RDB
|
rdb *rdb.RDB
|
||||||
cron *cron.Cron
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// New returns a new instance of Inspector.
|
// New returns a new instance of Inspector.
|
||||||
@@ -33,7 +31,6 @@ func NewInspector(r RedisConnOpt) *Inspector {
|
|||||||
}
|
}
|
||||||
return &Inspector{
|
return &Inspector{
|
||||||
rdb: rdb.NewRDB(c),
|
rdb: rdb.NewRDB(c),
|
||||||
cron: cron.New(cron.WithLocation(time.Local)),
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -799,36 +796,11 @@ type SchedulerEntry struct {
|
|||||||
// currently running schedulers.
|
// currently running schedulers.
|
||||||
func (i *Inspector) SchedulerEntries() ([]*SchedulerEntry, error) {
|
func (i *Inspector) SchedulerEntries() ([]*SchedulerEntry, error) {
|
||||||
var entries []*SchedulerEntry
|
var entries []*SchedulerEntry
|
||||||
fmt.Printf("start to record log file")
|
|
||||||
res, err := i.rdb.ListSchedulerEntries()
|
res, err := i.rdb.ListSchedulerEntries()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Errorf("ListSchedulerEntries err:%s", err.Error())
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
// 如果查询不到数据 从内存里面获取数据
|
|
||||||
if len(res) == 0 {
|
|
||||||
fmt.Println("search entries from mem")
|
|
||||||
for _, entry := range i.cron.Entries() {
|
|
||||||
job := entry.Job.(*enqueueJob)
|
|
||||||
var opts []Option
|
|
||||||
opt := stringifyOptions(job.opts)
|
|
||||||
for _, s := range opt {
|
|
||||||
if o, err := parseOption(s); err == nil {
|
|
||||||
// ignore bad data
|
|
||||||
opts = append(opts, o)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
entries = append(entries, &SchedulerEntry{
|
|
||||||
ID: job.id.String(),
|
|
||||||
Spec: job.cronspec,
|
|
||||||
Task: job.task,
|
|
||||||
Opts: opts,
|
|
||||||
Next: entry.Next,
|
|
||||||
Prev: entry.Prev,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
fmt.Printf("entries lens is %d \n", len(res))
|
|
||||||
} else {
|
|
||||||
for _, e := range res {
|
for _, e := range res {
|
||||||
task := NewTask(e.Type, e.Payload)
|
task := NewTask(e.Type, e.Payload)
|
||||||
var opts []Option
|
var opts []Option
|
||||||
@@ -847,7 +819,7 @@ func (i *Inspector) SchedulerEntries() ([]*SchedulerEntry, error) {
|
|||||||
Prev: e.Prev,
|
Prev: e.Prev,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
|
||||||
return entries, nil
|
return entries, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
25
scheduler.go
25
scheduler.go
@@ -297,3 +297,28 @@ func (s *Scheduler) clearHistory() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Scheduler) GetEntries() ([]*SchedulerEntry, error) {
|
||||||
|
var entries []*SchedulerEntry
|
||||||
|
|
||||||
|
for _, entry := range s.cron.Entries() {
|
||||||
|
job := entry.Job.(*enqueueJob)
|
||||||
|
var opts []Option
|
||||||
|
opt := stringifyOptions(job.opts)
|
||||||
|
for _, s := range opt {
|
||||||
|
if o, err := parseOption(s); err == nil {
|
||||||
|
// ignore bad data
|
||||||
|
opts = append(opts, o)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
entries = append(entries, &SchedulerEntry{
|
||||||
|
ID: job.id.String(),
|
||||||
|
Spec: job.cronspec,
|
||||||
|
Task: job.task,
|
||||||
|
Opts: opts,
|
||||||
|
Next: entry.Next,
|
||||||
|
Prev: entry.Prev,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return entries, nil
|
||||||
|
}
|
||||||
|
Reference in New Issue
Block a user