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

finish

Browse Source
Operations
Operations
Create branch
Create tag
Revert
Cherry-pick
main
git 2024-06-07 23:51:24 +08:00
parent 8be1a8ff3b
commit 4224490a66
Signed by: git
GPG Key ID: 3F65EFFA44207ADD
2 changed files with 14 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@ -49,4 +49,12 @@ func main() {
We welcome contributions in any form! If you have any issues, suggestions, or wish to add new features, let us know by creating an issue or pull request. We welcome contributions in any form! If you have any issues, suggestions, or wish to add new features, let us know by creating an issue or pull request.
## License ## License
machinery-plugins is based on the Apache 2.0 protocol. Please refer to [LICENSE](./LICENSE) for details. machinery-plugins is based on the Apache 2.0 protocol. Please refer to [LICENSE](./LICENSE) for details.
## 设计
push -> pending_tasks/{queue}/{task_uuid}
handle -> 获取keys-only, 判断是否有consumer，如果没有，原子添加consumer -> 成功 -> handler -> delete(每次全量key N，可以对吧lock)
delay -> delay_tasks/r{xx}-taskuuid
handlr -> 按key排序，获取最新的一个(可以大于当前时间) -> 原子put/delete操作 -> done (性能没有问题， 多个服务会有冲突,最多N-1，可以对比Lock)

7
brokers/etcd/etcd.go
View File

@ -202,13 +202,16 @@ func (b *etcdBroker) consumeOne(signature *Signature, taskProcessor iface.TaskPr
return nil return nil
} }
log.INFO.Printf("Task not registered with this worker. Requeuing message: %s", signature) log.INFO.Printf("Task not registered with this worker. Requeuing message: %s", signature)
b.requeueMessage(signature, taskProcessor) // b.requeueMessage(signature, taskProcessor)
return nil return nil
} }
log.DEBUG.Printf("Received new message: %s", signature) log.DEBUG.Printf("Received new message: %s", signature)
return taskProcessor.Process(signature.Signature) err := taskProcessor.Process(signature.Signature)
// ack
// b.deleteKey(si)
return err
} }
// StopConsuming 停止 // StopConsuming 停止