GoLang中后台作业终极指南

example
├── scheduler
│   ├── scheduler.go
├── processors
│   └── process.go
└── go.mod

// main file in scheduler package

package main
import (
 "github.com/gomodule/redigo/redis"
 "github.com/gocraft/work"
)
// 创建 redis 池
var redisPool = &redis.Pool{
 MaxActive: 5,
 MaxIdle: 5,
 Wait: true,
 Dial: func() (redis.Conn, error) {
  return redis.Dial("tcp", ":6379")
 },
}
//使用特定命名空间创建查询器
var enqueuer = work.NewEnqueuer("application_namespace", redisPool)

func main() {
 // 使用指定参数执行名为 "send_email "的任务。
 _, err := enqueuer.Enqueue("send_welcome_email", work.Q{"email_address": "test@example.com", "user_id": 4})
 if err != nil {
  log.Fatal(err)
 }
}

// main file in processor package

package main

import (
 "github.com/gomodule/redigo/redis"
 "github.com/gocraft/work"
 "os"
 "os/signal"
)
// Make a redis pool
var redisPool = &redis.Pool{
 MaxActive: 5,
 MaxIdle: 5,
 Wait: true,
 Dial: func() (redis.Conn, error) {
  return redis.Dial("tcp", ":6379")
 },
}
type Context struct{
    email string
    userId int64
}
func main() {
 // 创建一个新池。参数：
 // Context{} 是一个结构体，将作为请求的上下文。
 // 10 是最大并发量
 // "application_namespace" 是 Redis 命名空间
 // redisPool 是一个 Redis 池。
 pool := work.NewWorkerPool(Context{}, 10, "application_namespace", redisPool)

 // 添加将为每个作业执行的中间件
 pool.Middleware((*Context).Log)
 pool.Middleware((*Context).FindCustomer)

 // Map the name of jobs to handler functions
 pool.Job("send_welcome_email", (*Context).SendWelcomeEmail)

 // 开始处理工作
 pool.Start()

 // 等待退出信号：
 signalChan := make(chan os.Signal, 1)
 signal.Notify(signalChan, os.Interrupt, os.Kill)
 <-signalChan

 // Stop the pool
 pool.Stop()
}

func (c *Context) Log(job *work.Job, next work.NextMiddlewareFunc) error {
 fmt.Println("Starting job: ", job.Name)
 return next()
}


func (c *Context) FindCustomer(job *work.Job, next work.NextMiddlewareFunc) error {
 // 如果有 user_id 参数，则在上下文中设置该参数，以便将来的中间件和处理程序使用。
 if _, ok := job.Args["user_id"]; ok {
  c.userId = job.ArgInt64("user_id")
  c.email = job.ArgString("email_address")
  if err := job.ArgError(); err != nil {
   return err
  }
 }
 return next()
}


func (c *Context) SendWelcomeEmail(job *work.Job) error {
 // Extract arguments:
 addr := job.ArgString("email_address")
 if err := job.ArgError(); err != nil {
  return err
 }

 fmt.Println("Sending email")
 // 使用 net/smtp 库发送电子邮件
  // （电子邮件应使用 gmail 域名，密码应使用应用程序密码创建）
 from, password := "xyz@gmail.com", "sthkxlpixjferhfd"
 to := []string{
   addr,
 }

 smtpHost := "smtp.gmail.com"
 smtpPort := "587"
 message := []byte("This is a Welcome email message.")

 // 认证.
 auth := smtp.PlainAuth("", from, password, smtpHost)

 //发送电子邮件。
 err := smtp.SendMail(smtpHost+":"+smtpPort, auth, from, to, message)
 if err != nil {
  fmt.Println(err)
  return nil
 }

 fmt.Println("Email Sent Successfully!")
 return nil
}

enqueuer := work.NewEnqueuer("application_namespace", redisPool)
secondsInTheFuture := 300
_, err := enqueuer.EnqueueIn("send_welcome_email", secondsInTheFuture, work.Q{"email_address": "test@example.com"})

enqueuer := work.NewEnqueuer("application_namespace", redisPool)
job, err := enqueuer.EnqueueUnique("send_welcome_email", work.Q{"email_address": "xyz@gmail.com"}) // job returned
job, err = enqueuer.EnqueueUnique("send_welcome_email", work.Q{"email_address": "xyz@gmail.com"}) // job == nil -- this duplicate job isn't enqueued.
job, err = enqueuer.EnqueueUniqueIn("send_welcome_email", 300, work.Q{"email_address": "abc@gmail.com"}) // job != nil (diff email_address)

enqueuer := work.NewEnqueuer("application_namespace", redisPool)
job, err := enqueuer.EnqueueUniqueByKey("send_welcome_email", work.Q{"email_address": "xyz@gmail.com"}, map[string]interface{}{"my_key": "586"})
job, err = enqueuer.EnqueueUniqueInByKey("send_welcome_email", 120, work.Q{"email_address": "abc@gmail.com"}, map[string]interface{}{"my_key": "586"})

pool := work.NewWorkerPool(Context{}, 10, "application_namespace", redisPool)
pool.PeriodicallyEnqueue("0 * * * * *", "send_welcome_email") // 这将每分钟触发一次 "send_welcome_email "任务
pool.Job("send_welcome_email", (*Context).SendWelcomeEmail) // 仍然需要为这项工作单独注册一个处理程序