package main
import (
    "log"
    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)
const redisAddr = "127.0.0.1:6379"
func main() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: redisAddr},
        asynq.Config{
            // Specify how many concurrent workers to use
            Concurrency: 10,
            // Optionally specify multiple queues with different priority.
            Queues: map[string]int{
                "critical": 6,
                "default":  3,
                "low":      1,
            },
            // See the godoc for other configuration options
        },
    )
    // mux maps a type to a handler
    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.TypeEmailDelivery, tasks.HandleEmailDeliveryTask)
    mux.Handle(tasks.TypeImageResize, tasks.NewImageProcessor())
    // ...register other handlers...
    if err := srv.Run(mux); err != nil {
        log.Fatalf("could not run server: %v", err)
    }
}

package tasks
import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
    "github.com/hibiken/asynq"
)
// A list of task types.
const (
    TypeEmailDelivery   = "email:deliver"
    TypeImageResize     = "image:resize"
)
type EmailDeliveryPayload struct {
    UserID     int
    TemplateID string
}
type ImageResizePayload struct {
    SourceURL string
}
//----------------------------------------------
// Write a function NewXXXTask to create a task.
// A task consists of a type and a payload.
//----------------------------------------------
func NewEmailDeliveryTask(userID int, tmplID string) (*asynq.Task, error) {
    payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID})
    if err != nil {
        return nil, err
    }
    return asynq.NewTask(TypeEmailDelivery, payload), nil
}
func NewImageResizeTask(src string) (*asynq.Task, error) {
    payload, err := json.Marshal(ImageResizePayload{SourceURL: src})
    if err != nil {
        return nil, err
    }
    // task options can be passed to NewTask, which can be overridden at enqueue time.
    return asynq.NewTask(TypeImageResize, payload, asynq.MaxRetry(5), asynq.Timeout(20 * time.Minute)), nil
}
//---------------------------------------------------------------
// Write a function HandleXXXTask to handle the input task.
// Note that it satisfies the asynq.HandlerFunc interface.
//
// Handler doesn't need to be a function. You can define a type
// that satisfies asynq.Handler interface. See examples below.
//---------------------------------------------------------------
func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Sending Email to User: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // Email delivery code ...
    return nil
}
// ImageProcessor implements asynq.Handler interface.
type ImageProcessor struct {
    // ... fields for struct
}
func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Resizing image: src=%s", p.SourceURL)
    // Image resizing code ...
    return nil
}
func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

NewEmailDeliveryTask

NewImageResizeTask

//---------------------------------------------------------------
// Write a function HandleXXXTask to handle the input task.
// Note that it satisfies the asynq.HandlerFunc interface.
//
// Handler doesn't need to be a function. You can define a type
// that satisfies asynq.Handler interface. See examples below.
//---------------------------------------------------------------
func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Sending Email to User: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // Email delivery code ...
    return nil
}
// ImageProcessor implements asynq.Handler interface.
type ImageProcessor struct {
    // ... fields for struct
}
func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Resizing image: src=%s", p.SourceURL)
    // Image resizing code ...
    return nil
}
func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

package main
import (
    "log"
    "time"
    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)
const redisAddr = "127.0.0.1:6379"
func main() {
    client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
    defer client.Close()
    // ------------------------------------------------------
    // Example 1: Enqueue task to be processed immediately.
    //            Use (*Client).Enqueue method.
    // ------------------------------------------------------
    task, err := tasks.NewEmailDeliveryTask(42, "some:template:id")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err := client.Enqueue(task)
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
    // ------------------------------------------------------------
    // Example 2: Schedule task to be processed in the future.
    //            Use ProcessIn or ProcessAt option.
    // ------------------------------------------------------------
    info, err = client.Enqueue(task, asynq.ProcessIn(24*time.Hour))
    if err != nil {
        log.Fatalf("could not schedule task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
    // ----------------------------------------------------------------------------
    // Example 3: Set other options to tune task processing behavior.
    //            Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
    // ----------------------------------------------------------------------------
    task, err = tasks.NewImageResizeTask("https://example.com/myassets/image.jpg")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3 * time.Minute))
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
}

// FileBasedConfigProvider implements asynq.PeriodicTaskConfigProvider interface.
type FileBasedConfigProvider struct {
? ? ? ? filename string
}
type PeriodicTaskConfigContainer struct {
? ? ? ? Configs []*Config `yaml:"configs"`
}
type Config struct {
? ? ? ? Cronspec string `yaml:"cronspec"`
? ? ? ? TaskType string `yaml:"task_type"`
}  
// Parses the yaml file and return a list of PeriodicTaskConfigs.
func (p *FileBasedConfigProvider) GetConfigs() ([]*asynq.PeriodicTaskConfig, error) {
? ? ? ? data, err := os.ReadFile(p.filename)
? ? ? ? if err != nil {
? ? ? ? ? ? ? ? return nil, err
? ? ? ? }
? ? ? ? var c PeriodicTaskConfigContainer
? ? ? ? if err := yaml.Unmarshal(data, &c); err != nil {
? ? ? ? ? ? ? ? return nil, err
? ? ? ? }
? ? ? ? var configs []*asynq.PeriodicTaskConfig
? ? ? ? for _, cfg := range c.Configs {
? ? ? ? ? ? ? ? configs = append(configs, &asynq.PeriodicTaskConfig{Cronspec: cfg.Cronspec, Task: asynq.NewTask(cfg.TaskType, nil)})
? ? ? ? }
? ? ? ? return configs, nil
}

        provider := &FileBasedConfigProvider{filename: "./periodic_task_config.yml"}
? ? ? ? mgr, err := asynq.NewPeriodicTaskManager(
? ? ? ? ? ? ? ? asynq.PeriodicTaskManagerOpts{
? ? ? ? ? ? ? ? ? ? ? ? RedisConnOpt: asynq.RedisClientOpt{
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Addr: ? ? "127.0.0.1:6379",
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Password: "123456",
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? DB: ? ? ? 1,
? ? ? ? ? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? ? ? ? ? PeriodicTaskConfigProvider: provider, ? ? ? ? // this provider object is the interface to your config source
? ? ? ? ? ? ? ? ? ? ? ? SyncInterval: ? ? ? ? ? ? ? 10 * time.Second, // this field specifies how often sync should happen
? ? ? ? ? ? ? ? })
? ? ? ? if err != nil {
? ? ? ? ? ? ? ? log.Fatal(err)
? ? ? ? }
? ? ? ? if err := mgr.Run(); err != nil {
? ? ? ? ? ? ? ? log.Fatal(err)
? ? ? ? }

? ? ? ? loc, err := time.LoadLocation("Asia/Shanghai")
? ? ? ? if err != nil {
            panic(err)
        }
? ? ? ? scheduler := asynq.NewScheduler(
? ? ? ? ? ? ? asynq.RedisClientOpt{
? ? ? ? ? ? ? ? ? ? ? Addr: ? ? "127.0.0.1:6379",
? ? ? ? ? ? ? ? ? ? ? Password: "123456",
? ? ? ? ? ? ? ? ? ? ? DB: ? ? ? 1,
? ? ? ? ? ? ? },
? ? ? ? ? ? ? &asynq.SchedulerOpts{
? ? ? ? ? ? ? ? ? ? ? Location: loc,
? ? ? ? ? ? ? },
? ? ? ?  )
? ? ? ?  task := asynq.NewTask("example_task", nil)
? ? ? ?  // You can use cron spec string to specify the schedule.
? ? ? ?  entryID, err := scheduler.Register("*/1 * * * *", task)
? ? ? ?  if err != nil {
? ? ? ?  ? ? log.Fatal(err)
? ? ? ?  }
? ? ? ?  fmt.Println(entryID)
? ? ? ?  if err := scheduler.Run(); err != nil {
? ? ? ? ? ? ? log.Fatal(err)
? ? ? ?  }

pubsub := AsyncClient.Subscribe(context.Background(), "asynq:cancel")
	cancelCh := pubsub.Channel()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	now := time.Now()
	ch1 := make(chan int)
	stop := make(chan bool)
	go func(ctx context.Context, xxx string) {
		err := xxxx
		if err != nil {
			ch1 <- 2
			logx.Error(fmt.Sprintf("本次定時(shí)任務(wù)執(zhí)行失敗,失敗原因: %s", err))
			return
		}
		 for i := 0; i < 100; i++ {
		 	select {
		 	default:
		 		time.Sleep(1 * time.Second)
		 		logx.Info("我是任務(wù)主流程，我在運(yùn)行中...")
		 	case <-stop:
		 		logx.Info("收到退出信號(hào)，任務(wù)結(jié)束...")
		 		return
		 	}
		 }
		usedTime := time.Since(now)
		msg := fmt.Sprintf("本次定時(shí)任務(wù)成功結(jié)束執(zhí)行,用時(shí): %f秒", usedTime.Seconds())
		// Signal the goroutine has completed
		ch1 <- 1
	}(ctx, in.Args)
	for {
		select {
		case cd := <-ch1:
			if cd == 1 {
				return &CommonReply{Message: "本次定時(shí)任務(wù)成功結(jié)束執(zhí)行"}, nil
			} else {
				return &CommonReply{Message: "本次定時(shí)任務(wù)執(zhí)行失敗"}, nil
			}
		case msg := <-cancelCh:
			if msg.Payload == in.TaskId {
				// stop <- true
				respMsg := fmt.Sprintf("任務(wù)[%s]已被取消.Exist.", msg.Payload)
				logx.Info(respMsg)
				return &CommonReply{Message: respMsg}, nil
			}
		}
	}