common
/
machinery-plugins
Private
1
Fork 0
Code Pull Requests Releases Activity

finish handleDelayedTask

Browse Source
Operations
Operations
Create branch
Create tag
Revert
Cherry-pick
main
git 2024-06-08 09:57:35 +08:00
parent 73ffb245d5
commit 611441b5a2
Signed by: git
GPG Key ID: 3F65EFFA44207ADD
2 changed files with 10 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
brokers/etcd/delivery.go
View File

@ -26,24 +26,26 @@ type deliver struct {
signature *tasks.Signature signature *tasks.Signature
value []byte value []byte
key string key string
node string
aliveCancel func() aliveCancel func()
} }
func NewDelivery(ctx context.Context, client *clientv3.Client, key string) (Delivery, error) { func NewDelivery(ctx context.Context, client *clientv3.Client, key string, node string) (Delivery, error) {
d := &deliver{ d := &deliver{
ctx: ctx, ctx: ctx,
client: client, client: client,
key: key, key: key,
node: node,
} }
if err := d.assign(key); err != nil { if err := d.assign(key, node); err != nil {
return nil, err return nil, err
} }
return d, nil return d, nil
} }
func (d *deliver) assign(key string) error { func (d *deliver) assign(key string, node string) error {
assignKey := fmt.Sprintf("%s/assign", key) assignKey := fmt.Sprintf("%s/assign", key)
ctx, cancel := context.WithTimeout(d.ctx, time.Second*5) ctx, cancel := context.WithTimeout(d.ctx, time.Second*5)
defer cancel() defer cancel()
@ -54,7 +56,7 @@ func (d *deliver) assign(key string) error {
} }
cmp := clientv3.Compare(clientv3.CreateRevision(assignKey), "=", 0) cmp := clientv3.Compare(clientv3.CreateRevision(assignKey), "=", 0)
putReq := clientv3.OpPut(assignKey, "node", clientv3.WithLease(grantResp.ID)) putReq := clientv3.OpPut(assignKey, node, clientv3.WithLease(grantResp.ID))
getReq := clientv3.OpGet(key) getReq := clientv3.OpGet(key)
resp, err := d.client.Txn(ctx).If(cmp).Then(putReq, getReq).Commit() resp, err := d.client.Txn(ctx).If(cmp).Then(putReq, getReq).Commit()
if err != nil { if err != nil {

10
brokers/etcd/etcd.go
View File

@ -114,7 +114,7 @@ func (b *etcdBroker) StartConsuming(consumerTag string, concurrency int, taskPro
continue continue
} }
task, err := b.nextTask(getQueue(b.GetConfig(), taskProcessor)) task, err := b.nextTask(getQueue(b.GetConfig(), taskProcessor), consumerTag)
if err != nil { if err != nil {
if !errors.Is(err, haveNoTaskErr) { if !errors.Is(err, haveNoTaskErr) {
log.ERROR.Print(err) log.ERROR.Print(err)
@ -295,13 +295,13 @@ func (b *etcdBroker) GetDelayedTasks() ([]*tasks.Signature, error) {
return items, nil return items, nil
} }
func (b *etcdBroker) nextTask(queue string) (Delivery, error) { func (b *etcdBroker) nextTask(queue string, consumerTag string) (Delivery, error) {
for k := range b.keyMap { for k := range b.keyMap {
if !strings.Contains(k, queue) { if !strings.Contains(k, queue) {
continue continue
} }
d, err := NewDelivery(b.ctx, b.client, k) d, err := NewDelivery(b.ctx, b.client, k, consumerTag)
if err != nil { if err != nil {
continue continue
} }
@ -310,7 +310,7 @@ func (b *etcdBroker) nextTask(queue string) (Delivery, error) {
} }
time.Sleep(time.Second) time.Sleep(time.Second)
return b.nextTask(queue) return b.nextTask(queue, consumerTag)
} }
func (b *etcdBroker) listWatchTasks(ctx context.Context, queue string) error { func (b *etcdBroker) listWatchTasks(ctx context.Context, queue string) error {
@ -379,8 +379,6 @@ func (b *etcdBroker) handleDelayedTask(ctx context.Context) error {
} }
defer m.Unlock(ctx) // nolint defer m.Unlock(ctx) // nolint
log.INFO.Printf("acquired lock=%s, duration=%s", lockKey, ttl)
keyPrefix := "/machinery/v2/broker/delayed_tasks/eta-" keyPrefix := "/machinery/v2/broker/delayed_tasks/eta-"
end := strconv.FormatInt(time.Now().UnixMilli(), 10) end := strconv.FormatInt(time.Now().UnixMilli(), 10)
kvs, err := b.client.Get(b.ctx, keyPrefix+"0", clientv3.WithRange(keyPrefix+end)) kvs, err := b.client.Get(b.ctx, keyPrefix+"0", clientv3.WithRange(keyPrefix+end))