原文在這裡

在本教程中，我們將與佇列互動(dòng)並將其放入 Redis 伺服器
使用 github.com/hibiken/asynq 套件並為
建立調(diào)度程序 使用 github.com/robfig/cron 套件的排程任務(wù)。這一步一步
指南解釋如何設(shè)定隊(duì)列、安排任務(wù)以及優(yōu)雅地處理
關(guān)閉。

初始化模組

先為專案建立一個(gè)新的 Go 模組：

go mod init learn_queue_and_cron

創(chuàng)建 cron.go

cron.go 檔案負(fù)責(zé)在特定時(shí)間排程和執(zhí)行任務(wù)
間隔。下面是實(shí)作：

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/robfig/cron/v3"
)

func runCron(c *cron.Cron) {

    // Schedule a task to run every minute
    _, err := c.AddFunc("@every 1m", func() {
        fmt.Printf("Task executed every minute at: %v \n", time.Now().Local())
    })
    if err != nil {
        log.Fatal(err)
    }

    // Start the cron scheduler
    c.Start()
    log.Println("Cron scheduler started")

    // Keep the main goroutine running
    select {}
}

此程式碼安排任務(wù)每分鐘運(yùn)行一次並保持應(yīng)用程式運(yùn)行
確保調(diào)度程序連續(xù)工作。

創(chuàng)建queue.go

queue.go 檔案使用 Asynq 管理任務(wù)處理。程式碼如下：

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"

    "github.com/hibiken/asynq"
)

func runQueue(server *asynq.Server) {
    mux := asynq.NewServeMux()
    mux.HandleFunc("send_email", emailHandler)
    mux.HandleFunc("generate_report", reportHandler)

    if err := server.Run(mux); err != nil {
        log.Fatalf("Failed to run Asynq server: %v", err)
    }
}

func emailHandler(ctx context.Context, task *asynq.Task) error {
    var payload struct {
        To string `json:"to"`
    }
    if err := json.Unmarshal(task.Payload(), &payload); err != nil {
        return fmt.Errorf("failed to unmarshal payload: %w", err)
    }
    fmt.Printf("Sending email to: %s\n", payload.To)
    return nil
}

func reportHandler(ctx context.Context, task *asynq.Task) error {
    var payload struct {
        ReportID int `json:"report_id"`
    }
    if err := json.Unmarshal(task.Payload(), &payload); err != nil {
        return fmt.Errorf("failed to unmarshal payload: %w", err)
    }
    fmt.Printf("Generating report for ID: %d\n", payload.ReportID)
    return nil
}

解釋

• Handlers： emailHandler 和 reportHandler 透過解析處理任務(wù) 它們的有效負(fù)載並執(zhí)行相應(yīng)的操作。
• 任務(wù)佇列：定義了「send_email」和「generate_report」等任務(wù) 並透過Asynq的任務(wù)佇列進(jìn)行處理。

建立router.go

router.go 檔案設(shè)定 HTTP 端點(diǎn)來(lái)將任務(wù)排入佇列：

package main

import (
    "encoding/json"
    "net/http"

    "github.com/gin-gonic/gin"
    "github.com/hibiken/asynq"
)

func setupRouter(client *asynq.Client) *gin.Engine {
    r := gin.Default()

    r.POST("/enqueue/email", func(c *gin.Context) {
        var payload struct {
            To string `json:"to"`
        }
        if err := c.ShouldBindJSON(&payload); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body"})
            return
        }

        jsonPayload, err := json.Marshal(payload)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to marshal payload"})
            return
        }

        task := asynq.NewTask("send_email", jsonPayload)
        _, err = client.Enqueue(task)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to enqueue task"})
            return
        }

        c.JSON(http.StatusOK, gin.H{"message": "Email job enqueued"})
    })

    r.POST("/enqueue/report", func(c *gin.Context) {
        var payload struct {
            ReportID int `json:"report_id"`
        }
        if err := c.ShouldBindJSON(&payload); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body"})
            return
        }

        jsonPayload, err := json.Marshal(payload)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to marshal payload"})
            return
        }

        task := asynq.NewTask("generate_report", jsonPayload)
        _, err = client.Enqueue(task)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to enqueue task"})
            return
        }

        c.JSON(http.StatusOK, gin.H{"message": "Report job enqueued"})
    })

    return r
}

此程式碼使用 Gin 框架公開兩個(gè)端點(diǎn)以用於排隊(duì)任務(wù)。

建立main.go

main.go 檔案將所有內(nèi)容整合在一起：

package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/hibiken/asynq"
    "github.com/robfig/cron/v3"
)

func main() {
    c := cron.New()

    server := asynq.NewServer(
        asynq.RedisClientOpt{Addr: "localhost:6379"},
        asynq.Config{
            Concurrency: 10,
        },
    )

    client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
    defer client.Close()

    router := setupRouter(client)

    httpServer := &http.Server{
        Addr:    ":8080",
        Handler: router,
    }

    // Prepare shutdown context
    ctx, stop := context.WithCancel(context.Background())
    defer stop()
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt, syscall.SIGTERM)

    go runQueue(server)
    go runCron(c)
    go func() {
        if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Failed to run HTTP server: %v", err)
        }
    }()

    appShutdown(ctx, httpServer, c, server, quit)
}

func appShutdown(ctx context.Context, httpServer *http.Server, c *cron.Cron, server *asynq.Server, quit chan os.Signal) {
    // Wait for termination signal
    <-quit
    log.Println("Shutting down gracefully...")

    httpCtx, httpCancel := context.WithTimeout(ctx, 5*time.Second)
    defer httpCancel()
    if err := httpServer.Shutdown(httpCtx); err != nil {
        log.Printf("HTTP server shutdown error: %v", err)
    }

    server.Shutdown()
    c.Stop()

    log.Println("Application stopped")
}

此檔案結(jié)合了佇列、cron、HTTP 伺服器和關(guān)閉邏輯。

安裝依賴項(xiàng)

安裝所有必要的依賴項(xiàng)：

go mod tidy

建置並運(yùn)行應(yīng)用程式

使用以下命令建置並執(zhí)行應(yīng)用程式：

go build -o run *.go && ./run

測(cè)試應(yīng)用程式

存取以下端點(diǎn)將任務(wù)排隊(duì)：

• http://localhost:8080/enqueue/email
• http://localhost:8080/enqueue/report

觀察終端的任務(wù)執(zhí)行日誌。

規(guī)範(fàn)網(wǎng)址

欲了解更多詳細(xì)信息，請(qǐng)?jiān)L問我博客上的原始帖子。

以上是Go 中的 Redis 隊(duì)列和 Cron的詳細(xì)內(nèi)容。更多資訊請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！