大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章主要介绍“machinery中TaskProcessor的用法”,在日常操作中,相信很多人在machinery中TaskProcessor的用法问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”machinery中TaskProcessor的用法”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
创新互联从2013年成立,先为花山等服务建站,花山等地企业,进行企业商务咨询服务。为花山企业网站制作PC+手机+微官网三网同步一站式服务解决您的所有建站问题。
本文主要研究一下machinery的TaskProcessor
// TaskProcessor - can process a delivered task // This will probably always be a worker instance type TaskProcessor interface { Process(signature *tasks.Signature) error CustomQueue() string PreConsumeHandler() bool }
TaskProcessor接口定义了Process、CustomQueue、PreConsumeHandler方法
// Worker represents a single worker process type Worker struct { server *Server ConsumerTag string Concurrency int Queue string errorHandler func(err error) preTaskHandler func(*tasks.Signature) postTaskHandler func(*tasks.Signature) preConsumeHandler func(*Worker) bool } // CustomQueue returns Custom Queue of the running worker process func (worker *Worker) CustomQueue() string { return worker.Queue } // Process handles received tasks and triggers success/error callbacks func (worker *Worker) Process(signature *tasks.Signature) error { // If the task is not registered with this worker, do not continue // but only return nil as we do not want to restart the worker process if !worker.server.IsTaskRegistered(signature.Name) { return nil } taskFunc, err := worker.server.GetRegisteredTask(signature.Name) if err != nil { return nil } // Update task state to RECEIVED if err = worker.server.GetBackend().SetStateReceived(signature); err != nil { return fmt.Errorf("Set state to 'received' for task %s returned error: %s", signature.UUID, err) } // Prepare task for processing task, err := tasks.NewWithSignature(taskFunc, signature) // if this failed, it means the task is malformed, probably has invalid // signature, go directly to task failed without checking whether to retry if err != nil { worker.taskFailed(signature, err) return err } // try to extract trace span from headers and add it to the function context // so it can be used inside the function if it has context.Context as the first // argument. Start a new span if it isn't found. taskSpan := tracing.StartSpanFromHeaders(signature.Headers, signature.Name) tracing.AnnotateSpanWithSignatureInfo(taskSpan, signature) task.Context = opentracing.ContextWithSpan(task.Context, taskSpan) // Update task state to STARTED if err = worker.server.GetBackend().SetStateStarted(signature); err != nil { return fmt.Errorf("Set state to 'started' for task %s returned error: %s", signature.UUID, err) } //Run handler before the task is called if worker.preTaskHandler != nil { worker.preTaskHandler(signature) } //Defer run handler for the end of the task if worker.postTaskHandler != nil { defer worker.postTaskHandler(signature) } // Call the task results, err := task.Call() if err != nil { // If a tasks.ErrRetryTaskLater was returned from the task, // retry the task after specified duration retriableErr, ok := interface{}(err).(tasks.ErrRetryTaskLater) if ok { return worker.retryTaskIn(signature, retriableErr.RetryIn()) } // Otherwise, execute default retry logic based on signature.RetryCount // and signature.RetryTimeout values if signature.RetryCount > 0 { return worker.taskRetry(signature) } return worker.taskFailed(signature, err) } return worker.taskSucceeded(signature, results) } //SetPreConsumeHandler sets a custom handler for the end of a job func (worker *Worker) SetPreConsumeHandler(handler func(*Worker) bool) { worker.preConsumeHandler = handler }
Worker实现了TaskProcessor接口,其Process方法先通过worker.server.GetRegisteredTask获取taskFunc,然后通过signature更新state为RECEIVED,之后设置为STARTED,之后执行task.Call(),最后根据结果更新task为failed或者success
machinery的TaskProcessor接口定义了Process、CustomQueue、PreConsumeHandler方法。Worker实现了TaskProcessor接口,其Process方法先通过worker.server.GetRegisteredTask获取taskFunc,然后通过signature更新state为RECEIVED,之后设置为STARTED,之后执行task.Call(),最后根据结果更新task为failed或者success。
到此,关于“machinery中TaskProcessor的用法”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!