go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-09-20 02:55:54 +08:00
Code Issues Projects Releases Wiki Activity

[performance] Use BRPOPLPUSH if one queue is used

Browse Source
This commit is contained in:
Ken Hibino 2020-01-14 06:29:05 -08:00
parent f0251be5d2
commit 5a6f737589
2 changed files with 28 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
internal/rdb/rdb.go
View File

@ -60,11 +60,17 @@ func (r *RDB) Enqueue(msg *base.TaskMessage) error {
// is one and returns it. If all queues are empty, ErrNoProcessableTask
// error is returned.
func (r *RDB) Dequeue(qnames ...string) (*base.TaskMessage, error) {
var keys []string
for _, q := range qnames {
keys = append(keys, base.QueueKey(q))
var data string
var err error
if len(qnames) == 1 {
data, err = r.dequeueSingle(base.QueueKey(qnames[0]))
} else {
var keys []string
for _, q := range qnames {
keys = append(keys, base.QueueKey(q))
}
data, err = r.dequeue(keys...)
}
data, err := r.dequeue(keys...)
if err == redis.Nil {
return nil, ErrNoProcessableTask
}
@ -79,6 +85,11 @@ func (r *RDB) Dequeue(qnames ...string) (*base.TaskMessage, error) {
return &msg, nil
}
func (r *RDB) dequeueSingle(queue string) (data string, err error) {
// timeout needed to avoid blocking forever
return r.client.BRPopLPush(queue, base.InProgressQueue, time.Second).Result()
}
func (r *RDB) dequeue(queues ...string) (data string, err error) {
var args []interface{}
for _, qkey := range queues {

15
processor.go
View File

@ -124,8 +124,12 @@ func (p *processor) exec() {
msg, err := p.rdb.Dequeue(qnames...)
if err == rdb.ErrNoProcessableTask {
// queues are empty, this is a normal behavior.
// sleep to avoid slamming redis and let scheduler move tasks into queues.
time.Sleep(time.Second)
if len(p.queueConfig) > 1 {
// sleep to avoid slamming redis and let scheduler move tasks into queues.
// Note: With multiple queues, we are not using blocking pop operation and
// polling queues instead. This adds significant load to redis.
time.Sleep(time.Second)
}
return
}
if err != nil {
@ -221,6 +225,13 @@ func (p *processor) kill(msg *base.TaskMessage, e error) {
// If strict-priority is false, then the order of queue names are roughly based on
// the priority level but randomized in order to avoid starving low priority queues.
func (p *processor) queues() []string {
// skip the overhead of generating a list of queue names
// if we are processing one queue.
if len(p.queueConfig) == 1 {
for qname := range p.queueConfig {
return []string{qname}
}
}
if p.orderedQueues != nil {
return p.orderedQueues
}