.
├── main.go
├── sendtask.go
└── worker├── tasks.go└── worker.go// main.go 启动 worker
package mainimport "test/machinerydemo/worker"func main() {// 启动异步任务框架taskWorker := worker.NewAsyncTaskWorker(0)taskWorker.Launch()
}//sendtask.go, 测试发送任务
package mainimport ("context""test/machinerydemo/worker"
)func main() {worker.SendHelloWorldTask(context.Background())
}// worker/tasks.go 用来定义任务函数
package workerimport ("context""fmt""github.com/RichardKnop/machinery/v1/tasks"
)var (asyncTaskMap map[string]interface{}
)// 方法名
const (HelloWorldTaskName          = "HelloWorldTask"DeleteAppShareImageTaskName = "DeleteAppShareImageTask"
)// HelloWorld 测试异步任务
func HelloWorld() error {fmt.Println("Hello World!")return nil
}// SendHelloWorldTask 调用点调用此异步任务函数
func SendHelloWorldTask(ctx context.Context) {args := make([]tasks.Arg, 0)task, _ := tasks.NewSignature(HelloWorldTaskName, args)task.RetryCount = 5AsyncTaskCenter.SendTaskWithContext(ctx, task)
}func initAsyncTaskMap() {asyncTaskMap = make(map[string]interface{})asyncTaskMap[HelloWorldTaskName] = HelloWorld
}// worker/worker.go 用来定义后台 worker
package workerimport ("github.com/RichardKnop/machinery/v1"mchConf "github.com/RichardKnop/machinery/v1/config""github.com/RichardKnop/machinery/v1/log""github.com/RichardKnop/machinery/v1/tasks"
)var (AsyncTaskCenter *machinery.Server
)func init() {tc, err := NewTaskCenter()if err != nil {panic(err)}AsyncTaskCenter = tc
}func NewTaskCenter() (*machinery.Server, error) {cnf := &mchConf.Config{Broker:        "redis://localhost:6379",DefaultQueue:  "ServerTasksQueue",ResultBackend: "redis://localhost:6379",}// Create server instanceserver, err := machinery.NewServer(cnf)if err != nil {return nil, err}initAsyncTaskMap()return server, server.RegisterTasks(asyncTaskMap)
}func NewAsyncTaskWorker(concurrency int) *machinery.Worker {consumerTag := "TestWorker"// The second argument is a consumer tag// Ideally, each worker should have a unique tag (worker1, worker2 etc)worker := AsyncTaskCenter.NewWorker(consumerTag, concurrency)// Here we inject some custom code for error handling,// start and end of task hooks, useful for metrics for example.errorhandler := func(err error) {log.ERROR.Println("I am an error handler:", err)}pretaskhandler := func(signature *tasks.Signature) {log.INFO.Println("I am a start of task handler for:", signature.Name)}posttaskhandler := func(signature *tasks.Signature) {log.INFO.Println("I am an end of task handler for:", signature.Name)}worker.SetPostTaskHandler(posttaskhandler)worker.SetErrorHandler(errorhandler)worker.SetPreTaskHandler(pretaskhandler)return worker
}