How to use the task queue machinery of go language

This article "go language task queue machinery how to use" most people do not understand, so the editor summarized the following content, detailed, clear steps, with a certain reference value, I hope you can get something after reading this article, let's take a look at this "go language task queue machinery how to use" article.

Overview of use

Step 1: create server, configure parameters, and register task. (here server is only a configuration function, not a separate server process)

Step 2: start worker

Step 3: send task

It is exactly the same as the use of celery

Create serverfunc startServer () (* machinery.Server, error) {cnf: = & config.Config {Broker: "amqp://guest:guest@localhost:5672/", DefaultQueue: "machinery_tasks", ResultBackend: "amqp://guest:guest@localhost:5672/", ResultsExpireIn: 3600 / / Task validity period AMQP: & config.AMQPConfig {Exchange: "machinery_exchange", ExchangeType: "direct", BindingKey: "machinery_task", PrefetchCount: 3, / / limit consumption capacity} } / / Create server instance broker: = amqpbroker.New (cnf) backend: = amqpbackend.New (cnf) lock: = eagerlock.New () / / Task lock server: = machinery.NewServer (cnf, broker, backend, lock) / / Register tasks tasks: = mapping [string] interface {} {"add": exampletasks.Add "multiply": exampletasks.Multiply, "sum_ints": exampletasks.SumInts, "sum_floats": exampletasks.SumFloats, "concat": exampletasks.Concat, "split": exampletasks.Split, "panic_task": exampletasks.PanicTask "long_running_task": exampletasks.LongRunningTask,} return server, server.RegisterTasks (tasks)} create worker

Create a worker, and then you can start

Func worker () error {/ / Consumer tag consumerTag: = "machinery_worker" server, err: = startServer () if err! = nil {return err} / / the number of concurrency of the second parameter, 0 means unrestricted worker: = server.NewWorker (consumerTag) 0) / / Hook function errorhandler: = func (err error) {} pretaskhandler: = func (signature * tasks.Signature) {} posttaskhandler: = func (signature * tasks.Signature) {} worker.SetPostTaskHandler (posttaskhandler) worker.SetErrorHandler (errorhandler) worker.SetPreTaskHandler (pretaskhandler) return worker.Launch ()}

Startup result

INFO: 2021-05-01 08:28:27 worker.go:58 Launching a worker with the following settings:INFO: 2021-05-01 08:28:27 worker.go:59-Broker: amqp://192.168.120.101:5672INFO: 2021-05-01 08:28:27 worker.go:61-DefaultQueue: machinery_tasksINFO: 2021-05-01 08:28:27 worker.go:65-ResultBackend: amqp://192.168.120.101:5672INFO: 2021-05-01 08: 28:27 worker.go:67-AMQP: machinery_exchangeINFO: 08:28:27 on 2021-05-01 worker.go:68-Exchange: machinery_exchangeINFO: 2021-05-01 08:28:27 worker.go:69-ExchangeType: directINFO: 2021-05-01 08:28:27 worker.go:70-BindingKey: machinery_taskINFO: 2021-05-01 08:28:27 worker.go:71-PrefetchCount: 0INFO: 2021-05-01 08:28:27 amqp.go:96 [*] Waiting for messages. To exit press CTRL+C sends task server, _: = startServer () signature: = & tasks.Signature {Name: "add", Args: [] tasks.Arg {{Type: "int64", Value: 1,}, {Type: "int64", Value: 1,} asyncResult _: = server.SendTask (signature) fmt.Println (asyncResult.Get (time.Millisecond * 5)) / / wait interval In theory, the smaller the better / / asyncResult.GetWithTimeout (time.Second*120, time.Millisecond * 5) / / the first parameter is timeout and above is the content of this article on "how to use machinery in the task queue of go language". I believe we all have some understanding. I hope the content shared by the editor will be helpful to you. If you want to know more about the relevant knowledge, please follow the industry information channel.