大橙子网站建设,新征程启航

为企业提供网站建设、域名注册、服务器等服务

machinery的功能有哪些

本篇内容主要讲解“machinery的功能有哪些”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“machinery的功能有哪些”吧!

天全网站制作公司哪家好,找创新互联!从网页设计、网站建设、微信开发、APP开发、响应式网站等网站项目制作,到程序开发,运营维护。创新互联2013年至今到现在10年的时间,我们拥有了丰富的建站经验和运维经验,来保证我们的工作的顺利进行。专注于网站建设就选创新互联

特性

上面只是简单举了个例子,任务队列有着广泛的应用场景,比如大批量的计算任务,当有大量数据插入,通过拆分并分批插入任务队列,从而实现串行链式任务处理或者实现分组并行任务处理,提高系统鲁棒性,提高系统并发度;或者对数据进行预处理,定期的从后端存储将数据同步到到缓存系统,从而在查询请求发生时,直接去缓存系统中查询,提高查询请求的响应速度。适用任务队列的场景有很多,这里就不一一列举了。回归本文主题,既然我们要学习machinery,就要先了解一下他都有哪些特性呢?

  • 任务重试机制
  • 延迟任务支持
  • 任务回调机制
  • 任务结果记录
  • 支持Workflow模式:Chain,Group,Chord
  • 多Brokers支持:redis, AMQP, AWS SQS
  • 多Backends支持:Redis, Memcache, AMQP, MongoDB
 

架构

任务队列,简而言之就是一个放大的生产者消费者模型,用户请求会生成任务,任务生产者不断的向队列中插入任务,同时,队列的处理器程序充当消费者不断的消费任务。基于这种框架设计思想,我们来看下machinery的简单设计结构图例:

machinery的功能有哪些


 
  • Sender:业务推送模块,生成具体任务,可根据业务逻辑中,按交互进行拆分;
  • Broker:存储具体序列化后的任务,machinery中目前支持到Redis, AMQP,和SQS;
  • Worker:工作进程,负责消费者功能,处理具体的任务;
  • Backend:后端存储,用于存储任务执行状态的数据;
 

e.g

学习一门新东西,我都习惯先写一个demo,先学会了走,再学会跑。所以先来看一个例子,功能很简单,异步计算1到10的和。

先看一下配置文件代码:

broker: redis://localhost:6379

default_queue: "asong"

result_backend: redis://localhost:6379

redis:
  max_idle: 3
  max_active: 3
  max_idle_timeout: 240
  wait: true
  read_timeout: 15
  write_timeout: 15
  connect_timeout: 15
  normal_tasks_poll_period: 1000
  delayed_tasks_poll_period: 500
  delayed_tasks_key: "asong"
 

这里brokerresult_backend来实现。

主代码,完整版github获取:


func main()  {

 cnf,err := config.NewFromYaml("./config.yml",false)
 if err != nil{
  log.Println("config failed",err)
  return
 }

 server,err := machinery.NewServer(cnf)
 if err != nil{
  log.Println("start server failed",err)
  return
 }

 // 注册任务
 err = server.RegisterTask("sum",Sum)
 if err != nil{
  log.Println("reg task failed",err)
  return
 }

 worker := server.NewWorker("asong", 1)
 go func() {
  err = worker.Launch()
  if err != nil {
   log.Println("start worker error",err)
   return
  }
 }()

 //task signature
 signature := &tasks.Signature{
  Name: "sum",
  Args: []tasks.Arg{
   {
    Type:  "[]int64",
    Value: []int64{1,2,3,4,5,6,7,8,9,10},
   },
  },
 }

 asyncResult, err := server.SendTask(signature)
 if err != nil {
  log.Fatal(err)
 }
 res, err := asyncResult.Get(1)
 if err != nil {
  log.Fatal(err)
 }
 log.Printf("get res is %v\n", tasks.HumanReadableResults(res))

}
 

运行结果:

INFO: 2020/10/31 11:32:15 file.go:19 Successfully loaded config from file ./config.yml
INFO: 2020/10/31 11:32:15 worker.go:58 Launching a worker with the following settings:
INFO: 2020/10/31 11:32:15 worker.go:59 - Broker: redis://localhost:6379
INFO: 2020/10/31 11:32:15 worker.go:61 - DefaultQueue: asong
INFO: 2020/10/31 11:32:15 worker.go:65 - ResultBackend: redis://localhost:6379
INFO: 2020/10/31 11:32:15 redis.go:100 [*] Waiting for messages. To exit press CTRL+C
DEBUG: 2020/10/31 11:32:16 redis.go:342 Received new message: {"UUID":"task_9f01be1f-3237-49f1-8464-eecca2e50597","Name":"sum","RoutingKey":"asong","ETA":null,"GroupUUID":"","GroupTaskCount":0,"Args":[{"Name":"","Type":"[]int64","Value":[1,2,3,4,5,6,7,8,9,10]}],"Headers":{},"Priority":0,"Immutable":false,"RetryCount":0,"RetryTimeout":0,"OnSuccess":null,"OnError":null,"ChordCallback":null,"BrokerMessageGroupId":"","SQSReceiptHandle":"","StopTaskDeletionOnError":false,"IgnoreWhenTaskNotRegistered":false}
DEBUG: 2020/10/31 11:32:16 worker.go:261 Processed task task_9f01be1f-3237-49f1-8464-eecca2e50597. Results = 55
2020/10/31 11:32:16 get res is 55
 

好啦,现在我们开始讲一讲上面的代码流程,

  • 读取配置文件,这一步是为了配置     broker和     result_backend,这里我都选择的是     redis,因为电脑正好有这个环境,就直接用了。
  • Machinery 库必须在使用前实例化。实现方法是创建一个     Server实例。     Server是     Machinery配置和注册任务的基本对象。
  • 在你的     workders能消费一个任务前,你需要将它注册到服务器。这是通过给任务分配一个唯一的名称来实现的。
  • 为了消费任务,你需有有一个或多个worker正在运行。运行worker所需要的只是一个具有已注册任务的     Server实例。每个worker将只使用已注册的任务。对于队列中的每个任务,Worker.Process()方法将在一个goroutine中运行。可以使用     server.NewWorker的第二参数来限制并发运行的worker.Process()调用的数量(每个worker)。
  • 可以通过将     Signature实例传递给     Server实例来调用任务。
  • 调用     HumanReadableResults这个方法可以处理反射值,获取到最终的结果。
 

多功能

 

1. 延时任务

上面的代码只是一个简单machinery使用示例,其实machiney也支持延时任务的,可以通过在任务signature上设置ETA时间戳字段来延迟任务。

eta := time.Now().UTC().Add(time.Second * 20)
 signature.ETA = &eta
   

2. 重试任务

在将任务声明为失败之前,可以设置多次重试尝试。斐波那契序列将用于在一段时间内分隔重试请求。这里可以使用两种方法,第一种直接对tsak signature中的retryTimeoutRetryCount字段进行设置,就可以,重试时间将按照斐波那契数列进行叠加。

//task signature
 signature := &tasks.Signature{
  Name: "sum",
  Args: []tasks.Arg{
   {
    Type:  "[]int64",
    Value: []int64{1,2,3,4,5,6,7,8,9,10},
   },
  },
  RetryTimeout: 100,
  RetryCount: 3,
 }
 

或者,你可以使用return.tasks.ErrRetryTaskLater 返回任务并指定重试的持续时间。

func Sum(args []int64) (int64, error) {
 sum := int64(0)
 for _, arg := range args {
  sum += arg
 }

 return sum, tasks.NewErrRetryTaskLater("我说他错了", 4 * time.Second)

}
   

3. 工作流

上面我们讲的都是运行一个异步任务,但是我们往往做项目时,一个需求是需要多个异步任务以编排好的方式执行的,所以我们就可以使用machinery的工作流来完成。

 
3.1 Groups

Group 是一组任务,它们将相互独立地并行执行。还是画个图吧,这样看起来更明了:

machinery的功能有哪些

一起来看一个简单的例子:

 // group
 group,err :=tasks.NewGroup(signature1,signature2,signature3)
 if err != nil{
  log.Println("add group failed",err)
 }

 asyncResults, err :=server.SendGroupWithContext(context.Background(),group,10)
 if err != nil {
  log.Println(err)
 }
 for _, asyncResult := range asyncResults{
  results,err := asyncResult.Get(1)
  if err != nil{
   log.Println(err)
   continue
  }
  log.Printf(
   "%v  %v  %v\n",
   asyncResult.Signature.Args[0].Value,
   tasks.HumanReadableResults(results),
  )
 }
 

group中的任务是并行执行的。

 
3.2 chords

我们在做项目时,往往会有一些回调场景,machiney也为我们考虑到了这一点,Chord允许你定一个回调任务在groups中的所有任务执行结束后被执行。

machinery的功能有哪些

来看一段代码:

callback := &tasks.Signature{
  Name: "call",
 }



 group, err := tasks.NewGroup(signature1, signature2, signature3)
 if err != nil {

  log.Printf("Error creating group: %s", err.Error())
  return
 }

 chord, err := tasks.NewChord(group, callback)
 if err != nil {
  log.Printf("Error creating chord: %s", err)
  return
 }

 chordAsyncResult, err := server.SendChordWithContext(context.Background(), chord, 0)
 if err != nil {
  log.Printf("Could not send chord: %s", err.Error())
  return
 }

 results, err := chordAsyncResult.Get(time.Duration(time.Millisecond * 5))
 if err != nil {
  log.Printf("Getting chord result failed with error: %s", err.Error())
  return
 }
 log.Printf("%v\n", tasks.HumanReadableResults(results))
 

上面的例子并行执行task1、task2、task3,聚合它们的结果并将它们传递给callback任务。

 
3.3 chains

chain就是一个接一个执行的任务集,每个成功的任务都会触发chain中的下一个任务。

machinery的功能有哪些

看这样一段代码:

//chain
 chain,err := tasks.NewChain(signature1,signature2,signature3,callback)
 if err != nil {

  log.Printf("Error creating group: %s", err.Error())
  return
 }
 chainAsyncResult, err := server.SendChainWithContext(context.Background(), chain)
 if err != nil {
  log.Printf("Could not send chain: %s", err.Error())
  return
 }

 results, err := chainAsyncResult.Get(time.Duration(time.Millisecond * 5))
 if err != nil {
  log.Printf("Getting chain result failed with error: %s", err.Error())
 }
 log.Printf(" %v\n", tasks.HumanReadableResults(results))

上面的例子执行task1,然后是task2,然后是task3。当一个任务成功完成时,结果被附加到chain中下一个任务的参数列表的末尾,最终执行callback任务。

到此,相信大家对“machinery的功能有哪些”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!


当前题目:machinery的功能有哪些
转载注明:http://dzwzjz.com/article/ijooci.html
在线咨询
服务热线
服务热线:028-86922220
TOP